# Step 4: Write Code That Subscribes to an Event Channel

Follow instructions in this step to build a subscriber client.

:::note
When you copy the code snippets in this page, ensure that the indentation starts from the comment block below the `with grpc.secure_channel` statement that you added in [Step 2: Build the Python Client](/docs/platform/pub-sub-api/guide/qs-build-python-client.md). In Python, code blocks are delimited by their indentation.
:::

1. Get the topic name that you want to subscribe to.

   The topic format is:

   - For a custom platform event: `/event/EventName__e`
   - For a standard platform event: `/event/EventName`
   - For a change data capture channel that captures events for all selected entities: `/data/ChangeEvents`
   - For a change data capture single-entity channel for a standard object: `/data/{StandardObjectName}ChangeEvent`
   - For a change data capture single-entity channel for a custom object: `/data/{CustomObjectName}__ChangeEvent`
   - For a change data capture custom channel: `/data/CustomChannelName__chn`
   - For a custom platform event channel: `/event/CustomChannelName__chn`

2. Create a generator function to make a FetchRequest stream. After the semaphore is released in the client, the `fetchReqStream` function acquires a lock and sends a FetchRequest. In this FetchRequest, `num_requested` is the number of events that the client requests from the server. The specified `num_requested` of events can be sent in one or more FetchResponses to the client, with each FetchResponse containing a batch of events. In this case, we set it to 1.

   ```python
   def fetchReqStream(topic):
       while True:
           semaphore.acquire()
           yield pb2.FetchRequest(
               topic_name = topic,
               replay_preset = pb2.ReplayPreset.LATEST,
               num_requested = 1)
   ```

3. Create a decoding function to decode the payloads of received event messages.

   ```python
   def decode(schema, payload):
     schema = avro.schema.parse(schema)
     buf = io.BytesIO(payload)
     decoder = avro.io.BinaryDecoder(buf)
     reader = avro.io.DatumReader(schema)
     ret = reader.read(decoder)
     return ret
   ```

4. Make the subscribe call, and handle received event messages. Decode the payloads of the events with your decoding function. Store the latest replay ID received. You can use the replay ID later to restart a subscription after the last consumed event, if necessary. For more information, see [Replaying an Event Stream](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/subscribe-rpc.md#replaying-an-event-stream).

   ```python
   mysubtopic = "/data/OpportunityChangeEvent"
   print('Subscribing to ' + mysubtopic)
   substream = stub.Subscribe(fetchReqStream(mysubtopic),
            metadata=authmetadata)
   for event in substream:
     if event.events:
       semaphore.release()
       print("Number of events received: ", len(event.events))
       payloadbytes = event.events[0].event.payload
       schemaid = event.events[0].event.schema_id
       schema = stub.GetSchema(
                pb2.SchemaRequest(schema_id=schemaid),
            	metadata=authmetadata).schema_json
       decoded = decode(schema, payloadbytes)
       print("Got an event!", json.dumps(decoded))
     else:
       print("[", time.strftime('%b %d, %Y %l:%M%p %Z'),
       "] The subscription is active.")
     latest_replay_id = event.latest_replay_id
   ```

:::note
In this example, the `event` variable represents a FetchResponse. The code example processes one event at a time. When a FetchResponse containing an event is received, the semaphore is released, which results in the `fetchReqStream` function locking the semaphore and sending a new FetchRequest. The new FetchRequest requests one event in `num_requested`. This flow control works only when requesting and processing one event at a time. For more information about flow control, see [Pull Subscription and Flow Control](/docs/platform/pub-sub-api/guide/flow-control.md). For more information about semaphores, see [Semaphore Objects](https://docs.python.org/3/library/threading.html#semaphore-objects) in the Python documentation.
:::

If you run your code at this point, you don’t receive any event messages unless you or Salesforce publishes an event message.

For change data capture events, make sure that Change Data Capture is tracking the object by selecting the object in Setup, on the Change Data Capture page. For this example, select the Opportunity object. Then make a change to an opportunity so that Salesforce generates an event message.

Salesforce publishes most standard platform events, including real-time event monitoring events, in response to an action in Salesforce. You can publish only the standard events that support the `create()` call. For more information, see [Standard Platform Event Object List](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_objects_list.htm) in the [*Platform Events Developer Guide*](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_intro.htm).

## Received Event Example

If you subscribe to the `/data/OpportunityChangeEvent` topic and make a change to an opportunity, you receive a change event similar to this event message. This event message is for an opportunity whose `Amount` field was changed and the `Type` field was cleared (set to null).

:::note
`ChangeEventHeader` contains bitmap fields that you must decode before reading their contents. These fields are: `nulledFields`, `changedFields`, and `diffFields`. For more information, see [Event Deserialization Considerations](/docs/platform/pub-sub-api/guide/event-deserialization-considerations.md).
:::

```json
{
  "ChangeEventHeader": {
    "entityName": "Opportunity",
    "recordIds": ["006R0000002FyVgIAK"],
    "changeType": "UPDATE",
    "changeOrigin": "com/salesforce/api/soap/55.0;client=SfdcInternalAPI/",
    "transactionKey": "00006230-d3c7-f430-989e-f686e0b451bf",
    "sequenceNumber": 1,
    "commitTimestamp": 1656432588000,
    "commitNumber": 77385917926,
    "commitUser": "005R0000000Iu3SIAS",
    "nulledFields": ["0x0800"],
    "diffFields": [],
    "changedFields": ["0x21000840"]
  },
  "AccountId": null,
  "IsPrivate": null,
  "Name": null,
  "Description": null,
  "StageName": null,
  "Amount": 110000,
  "Probability": null,
  "ExpectedRevenue": null,
  "TotalOpportunityQuantity": null,
  "CloseDate": null,
  "Type": null,
  "NextStep": null,
  "LeadSource": null,
  "IsClosed": null,
  "IsWon": null,
  "ForecastCategory": null,
  "ForecastCategoryName": null,
  "CampaignId": null,
  "HasOpportunityLineItem": null,
  "Pricebook2Id": null,
  "OwnerId": null,
  "CreatedDate": null,
  "CreatedById": null,
  "LastModifiedDate": 1656432588000,
  "LastModifiedById": null,
  "LastStageChangeDate": null,
  "ContactId": null,
  "ContractId": null,
  "LastAmountChangedHistoryId": "008R0000003RnLUIA0",
  "LastCloseDateChangedHistoryId": null
}
```

You can also publish a custom platform event. The next step shows you how to do that using the Pub/Sub API in Python.

**See Also**

- [Subscription Channels](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_subscribe_channels.htm)
- [Group Platform Events into One Stream with a Custom Channel](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_subscribe_custom_channels.htm)