Consuming Messages from a Stream

Consume messages from a stream in the Streaming service.

To review requirements for consuming messages, see Consuming Messages.

You can't use the Console to consume messages, but you can use it to show the recent messages in a stream.
1. Access the Streaming service in the Console.
2. On the Streams page, click the name of the stream for which you want to see recent messages.
3. On the stream details page, under Recent Messages, click Load Messages.
Note

Streams that use private endpoints aren't accessible from the internet, so their messages aren't displayed in the Console.

For information about using the CLI, see Command Line Interface (CLI). For a complete list of flags and options available for CLI commands, see the Command Line Reference.

To create a cursor

oci streaming stream cursor create-cursor --stream-id <stream_OCID> --partition <partition> --type <cursor_type> --endpoint <messages_endpoint>

For example:

oci streaming stream cursor create-cursor --stream-id ocid1.stream.oc1.phx.exampleuniqueID --partition 0 --type TRIM_HORIZON --endpoint https://cell-1.streaming.us-phoenix-1.oci.oraclecloud.com
{
  "data": {
    "value": "examplecursorvalue"
  }
}

To create a group cursor

oci streaming stream cursor create-group-cursor --stream-id <stream_OCID> --group-name <consumer_group_name> --type <cursor_type> --endpoint <messages_endpoint>

For example:

oci streaming stream cursor create-group-cursor --stream-id ocid1.stream.oc1.phx.exampleuniqueID --group-name MyConsumerGroup --type TRIM_HORIZON --endpoint https://cell-1.streaming.us-phoenix-1.oci.oraclecloud.com
{
  "data": {
    "value": "examplegroupcursorvalue"
  }
}

To get messages

oci streaming stream message get --stream-id <stream_OCID> --cursor <cursor> --endpoint <messages_endpoint>

Your first request to get messages should use the value returned when you created a cursor. Each subsequent request should use the opc-next-cursor value returned in the previous response.

For example:

oci streaming stream message get --stream-id ocid1.stream.oc1.phx.exampleuniqueID --cursor examplecursorvalue --endpoint https://cell-1.streaming.us-phoenix-1.oci.oraclecloud.com
{
  "data": [
    {
      "key": "a2V5MQ==",
      "offset": 0,
      "partition": "0",
      "stream": "MyStream",
      "timestamp": "2020-11-03T21:52:58.470000+00:00",
      "value": "dmFsdWUx"
    },
    {
      "key": "a2V5Mg==",
      "offset": 1,
      "partition": "0",
      "stream": "MyStream",
      "timestamp": "2020-11-03T21:52:58.470000+00:00",
      "value": "dmFsdWUy"
    },
    {
      "key": "a2V5MQ==",
      "offset": 2,
      "partition": "0",
      "stream": "MyStream",
      "timestamp": "2020-11-03T22:00:48.305000+00:00",
      "value": "dmFsdWUx"
    },
    {
      "key": "a2V5Mg==",
      "offset": 3,
      "partition": "0",
      "stream": "MyStream",
      "timestamp": "2020-11-03T22:00:48.305000+00:00",
      "value": "dmFsdWUy"
    }
  ],
  "opc-next-cursor": "examplenextcursorvalue"
}

To send a heartbeat

oci streaming stream group heartbeat --stream-id <stream_OCID> --cursor <cursor> --endpoint <messages_endpoint>

Your first heartbeat request should use the value returned when you created a group cursor. Each subsequent request should use the value returned in the previous response.

For example:

oci streaming stream group heartbeat --stream-id ocid1.stream.oc1.phx.exampleuniqueID --cursor examplecursorvalue --endpoint https://cell-1.streaming.us-phoenix-1.oci.oraclecloud.com
{
  "data": {
    "value": "examplenextcursorvalue"
  }
}

For information about using the API and signing requests, see REST API documentation and Security Credentials. For information about SDKs, see SDKs and the CLI.

Use the following API operations to consume messages:

Oracle Cloud Infrastructure Documentation

Consuming Messages from a Stream