Event streaming

CPMS publishes what happens at your charging stations and on your roaming connections to a message broker you consume from in real time. Use it when you need station- or roaming-level data as it happens — connector status, transactions, meter values, OCPI pushes and commands — at a volume and latency that polling the REST API or Webhooks cannot serve.

There are three streams:

Which one do I need?

• Webhooks — you want the CPMS view of a session or CDR (SessionDto, CdrDto), pushed over HTTP.
• Business events — you want station events in one shape, whatever OCPP version the station speaks.
• Raw OCPP — you want full fidelity: every action, every field, CPMS commands and station replies. You parse OCPP yourself.
• Raw OCPI — you want to audit or mirror what CPMS exchanges with eMSPs and hubs: location and session pushes, CDRs, tokens, commands.

flowchart LR
    CS[Charging station] <-->|OCPP| CPMS
    P[Roaming partner] <-->|OCPI| CPMS
    CPMS -->|ocpp-raw| RI[OCPPRawInbound]
    CPMS -->|ocpp-raw| RO[OCPPRawOutbound]
    CPMS -->|OcpiInbound / OcpiOutbound| OI[OCPI inbound<br/>OCPI outbound]
    CPMS -->|cpo-event| BE[ConnectorStatus<br/>ChargingSession<br/>MeterValues<br/>EVChargingNeeds]
    RI --> You[Your consumer]
    RO --> You
    OI --> You
    BE --> You

Transports

Each category is published to its own Kafka topic, SNS topic or SQS queue.

Use SNS when you want to fan out to several subscribers (SQS, Lambda, HTTPS); use SQS FIFO when you need per-station order without running Kafka.

Getting access

Streaming is enabled per environment by Solidstudio; it is not configured through the Public API. Agree the following with your Solidstudio contact:

• Transport — Kafka, SNS or SQS.
• Destinations — Kafka topic names, SNS topic ARNs or SQS queue URLs, one per category you want. Kafka topic names for OCPP and business events default to the category names on this page; raw OCPI topic names (for example raw-ocpi-inbound / raw-ocpi-outbound) are agreed during onboarding.
• Credentials — for Kafka, the broker addresses and either a SCRAM username/password or, for raw OCPI, optionally a client certificate for mutual TLS. For SNS/SQS in your own AWS account, the topic or queue access policy must allow Solidstudio's publisher role to call sns:Publish / sqs:SendMessage; we share the role ARN during onboarding.

Envelope

Every raw OCPP message and business event is a JSON object with these fields (raw OCPI messages use their own envelope):

Treat the envelope as open: ignore fields you don't recognise. New fields and new categories may be added without notice.

Raw OCPP messages

Additional fields:

{
  "category": "OCPPRawInbound",
  "chargingStationId": "ACE00007",
  "type": "ocpp-raw",
  "connectionId": "740ac88a-6be6-407a-a2c5-67b51964cb6d",
  "raw": "[2,\"c0b5e4d2-1f3a-4a8e-9a77-2b1f0f3d9e10\",\"StatusNotification\",{\"connectorId\":1,\"errorCode\":\"NoError\",\"status\":\"Available\"}]",
  "ts": "2025-01-01T00:00:00.000Z",
  "metadata": {
    "source": "Solidstudio",
    "publishedAt": "2025-01-01T00:00:00.012Z"
  }
}

Reading raw

raw is an OCPP frame:

A request and its reply share the messageId and travel on opposite topics: a station's CALL is on OCPPRawInbound and the CPMS reply on OCPPRawOutbound, and the other way round for CPMS commands. The two topics are not ordered relative to each other — join them on chargingStationId + messageId.

Commands sent through the v2 commands API return an ocppMessageId. It is the messageId of the CALL on OCPPRawOutbound, so you can follow a command you sent to the station's reply.

Transport differences

SNS message attributes

When present, these attributes let you filter in the SNS subscription instead of in your consumer:

Example filter policy — only faulted connectors:

{ "action": ["StatusNotification"], "status": ["Faulted"] }

Raw OCPI messages

Every state-changing OCPI call between CPMS and a roaming partner (eMSP or hub) is published as one message containing both the request and the response.

Inbound and outbound go to separate topics. On Kafka, the message key is the CPMS identifier of the OCPI partner, so one partner's traffic stays in order within a topic.

{
  "type": "OcpiOutbound",
  "protocolVersion": "2.2.1",
  "tracing": {
    "id": "3f6c2a1e-8b7d-4c55-9f1a-0d2e6b7c8a91",
    "correlationId": "3f6c2a1e-8b7d-4c55-9f1a-0d2e6b7c8a91",
    "causationId": "3f6c2a1e-8b7d-4c55-9f1a-0d2e6b7c8a91",
    "time": "2025-01-01T00:00:00.250Z",
    "source": "Solidstudio"
  },
  "request": {
    "method": "PATCH",
    "url": "https://emsp.example.com/ocpi/emsp/2.2.1/locations/PL/SOL/LOC1/PL*SOL*E1",
    "headers": {
      "X-Request-ID": "6b1f8e0c-2d4a-4f3b-8c9e-1a2b3c4d5e6f",
      "X-Correlation-ID": "a7e3c9d1-5b2f-4e8a-9c0d-7f6e5d4c3b2a"
    },
    "body": { "status": "CHARGING", "last_updated": "2025-01-01T00:00:00Z" }
  },
  "response": {
    "headers": {
      "X-Request-ID": "6b1f8e0c-2d4a-4f3b-8c9e-1a2b3c4d5e6f",
      "X-Correlation-ID": "a7e3c9d1-5b2f-4e8a-9c0d-7f6e5d4c3b2a"
    },
    "body": { "status_code": 1000, "status_message": "Success", "timestamp": "2025-01-01T00:00:00.200Z" }
  },
  "version": "1.0"
}

Notes:

• Read-only calls are not streamed by default. GET, HEAD, OPTIONS and TRACE are filtered out, so pulls (a partner paginating GET /locations, CPMS fetching tokens) do not appear; pushes, commands and authorisation calls do.
• No HTTP status code. Judge the outcome from response.body.status_code (OCPI 1xxx success, 2xxx client error, 3xxx server error).
• Credentials are never streamed. Authorization, Cookie and similar headers are dropped or replaced with ....
• Non-JSON bodies are wrapped as { "body": "<text>" }.
• Correlate with partner logs on X-Request-ID / X-Correlation-ID — they are the values sent on the wire.

Business events

Business events are published for every OCPP version, as long as the station sends the underlying message.

Each event carries the envelope plus a payload object specific to the category.

ConnectorStatus

{
  "category": "ConnectorStatus",
  "chargingStationId": "ACE00007",
  "type": "cpo-event",
  "connectionId": "740ac88a-6be6-407a-a2c5-67b51964cb6d",
  "ts": "2025-01-01T00:00:00Z",
  "payload": {
    "connectorId": 1,
    "evseId": 1,
    "status": "Available"
  },
  "metadata": {
    "source": "Solidstudio",
    "publishedAt": "2025-01-01T00:00:00Z"
  }
}

ChargingSession

One event when the transaction starts (Started) and one when it ends (Ended).

Transaction started:

{
  "category": "ChargingSession",
  "chargingStationId": "ACE00007",
  "type": "cpo-event",
  "connectionId": "740ac88a-6be6-407a-a2c5-67b51964cb6d",
  "ts": "2025-01-01T00:00:00Z",
  "payload": {
    "event": "Started",
    "connectorId": 1,
    "evseId": 1,
    "ocppTransactionId": 12345,
    "idTag": "DEADBEEF",
    "meterStart": 0
  },
  "metadata": {
    "source": "Solidstudio",
    "publishedAt": "2025-01-01T00:00:00Z"
  }
}

Transaction ended:

{
  "category": "ChargingSession",
  "chargingStationId": "ACE00007",
  "type": "cpo-event",
  "connectionId": "740ac88a-6be6-407a-a2c5-67b51964cb6d",
  "ts": "2025-01-01T00:00:00Z",
  "payload": {
    "event": "Ended",
    "connectorId": 1,
    "evseId": 1,
    "ocppTransactionId": 12345,
    "idTag": "DEADBEEF",
    "meterStart": 0,
    "meterStop": 100,
    "reason": "Remote"
  },
  "metadata": {
    "source": "Solidstudio",
    "publishedAt": "2025-01-01T00:00:00Z"
  }
}

MeterValues

Units are passed through as the station reports them; normalise before aggregating across stations.

{
  "category": "MeterValues",
  "chargingStationId": "ACE00007",
  "type": "cpo-event",
  "connectionId": "740ac88a-6be6-407a-a2c5-67b51964cb6d",
  "ts": "2025-01-01T00:00:00Z",
  "payload": {
    "connectorId": 1,
    "evseId": 1,
    "meterValue": [
      {
        "timestamp": "2025-01-01T00:00:00Z",
        "sampledValue": [
          { "value": 32, "measurand": "Power.Active.Import", "unit": "kW" },
          { "value": 100, "measurand": "Energy.Active.Import.Register", "unit": "kWh" }
        ]
      }
    ]
  },
  "metadata": {
    "source": "Solidstudio",
    "publishedAt": "2025-01-01T00:00:00Z"
  }
}

EVChargingNeeds

Only published for stations on OCPP 2.1 or OCPP 1.6 with the ISO 15118 (V2G) extension.

{
  "category": "EVChargingNeeds",
  "chargingStationId": "ACE00007",
  "type": "cpo-event",
  "connectionId": "740ac88a-6be6-407a-a2c5-67b51964cb6d",
  "ts": "2025-01-01T00:00:00Z",
  "payload": {
    "evseId": 1,
    "timestamp": "2025-01-01T00:00:00Z",
    "chargingNeeds": {
      "mobilityNeedsMode": "EVCC",
      "requestedEnergyTransfer": "AC_three_phase_BPT",
      "controlMode": "Dynamic",
      "departureTime": "2026-01-01T00:00:00Z",
      "v2xChargingParameters": {
        "evMaxEnergyRequest": 9300,
        "evTargetEnergyRequest": 4800,
        "maxDischargePower": -11000,
        "minChargePower": 0,
        "evMinEnergyRequest": -32800,
        "maxChargePower": 11000,
        "minDischargePower": 0
      }
    }
  },
  "metadata": {
    "source": "Solidstudio",
    "publishedAt": "2025-01-01T00:00:00Z"
  }
}

Semantics you must design for

• Out of band. Events are published asynchronously, after CPMS has handled the OCPP message or OCPI call. A slow or unavailable broker never delays the station or the partner — and an event can therefore arrive after the REST API already reflects the change.
• At-most-once, with possible duplicates. A publish that keeps failing is logged and dropped; some publishers retry a few times first, which can produce a duplicate. Deduplicate raw OCPI messages on tracing.id. Raw OCPP messages and business events have no event ID: deduplicate on natural keys — chargingStationId + messageId from raw for raw OCPP, chargingStationId + category + ts + the payload's identifiers for business events.
• Order by timestamp, not by arrival. Use ts, or tracing.time for raw OCPI. Arrival order only holds per station (OCPP) or per partner (OCPI) on Kafka, per station on SQS FIFO, and never across topics.
• Reconnects change connectionId. Use it to group a connection's traffic, not to identify a station.
• Reconcile from the REST API. For completeness, backfill from GET /api/v1/charging-stations/{chargingStationId}/ocpp-logs, POST /api/v1/sessions/rsql-list, GET /api/v1/connectors and POST /api/v1/cdrs/rsql-list.

Related

• Webhooks — session and CDR updates over HTTP
• Remote commands — ocppMessageId for correlating commands with raw OCPP traffic
• Session flows — how OCPP transactions become CPMS sessions
• OCPI partners — the roaming partners whose traffic appears on the raw OCPI stream