Welcome

This documentation was last updated on Mon Apr 22 08:00:15 UTC 2024 and covers Mambu Version v9.160.2

Welcome to the Mambu Streaming API Reference documentation!

The Streaming API allows you to subscribe to constant, real-time updates of activities within Mambu. If your connection is interrupted, your cursor position is saved so that when you reconnect, you will pick up exactly where you left off. The Streaming API is independent from the Mambu API and webhook infrastructure.

For information about setting up event streaming notification templates in the Mambu UI, see Streaming API in our User Guide.

About Mambu Streaming API

The purpose of the Streaming API is to provide a high-performance and reliable mechanism to stream significant amounts of data out of Mambu on a constant basis, without posing unnecessary pressure on the API and webhook infrastructure.

In order to subscribe to Mambu notifications events and consume them, a dedicated API is exposed to be used by clients.

The typical workflow is:

Create a subscription specifying the topic names you want to read.
Start reading batches of events from the subscription.
Commit the cursors found in the event batches back to Mambu, which will store the offsets.

If the connection is closed, and later restarted, clients will get events from the point of your last cursor commit. If you need more than one client for your subscription to distribute the load you can read the subscription with multiple clients, for more details please check the get events endpoint.

Base URLs

The base URL for requests to the Streaming API is found below:

https://TENANT_NAME.mambu.com/api/v1

To make requests to your tenant's sandbox, use the following base URL:

https://TENANT_NAME.sandbox.mambu.com/api/v1

Replace TENANT_NAME above with the tenant name in your Mambu tenant URL.

HTTP Verbs

Standard HTTP verbs are used to indicate the API request method.

Verb	Function
`GET`	To get events or statistics from a subscription
`POST`	To create a subscription or commit a cursor
`DELETE`	To delete a subscription

Authentication

To access the Streaming API, you must create an API consumer that has the Manage Events Streaming (MANAGE_EVENTS_STREAMING) permission assigned to it and create API keys using this API consumer.

API keys inherit the scope of access settings from the API consumer that creates them. You must authenticate every request to the Streaming API using an API key in the request header.

You may create and manage API consumers and keys either through the Mambu UI or using the API Consumers endpoint provided by Mambu API v2. For more information, see API Consumers in our User Guide.

Example authenticated request using cURL

curl  --request GET 'https://TENANT_NAME.mambu.com/api/v1/subscriptions/d45a34ed341321bca4d89e42452dc074/events' \
--header 'apikey: i9TCzwUBwyTVQrfPEAhk0oEpOUCt0O2M'

Open API Specification

Use this link to access the latest Open API Specification.

This documentation is automatically generated from an OpenAPI Specification (OAS).

This OAS file can also be used to quickly scaffold client SDKs, in various languages, for interacting with our APIs using a number of tools, including the online Swagger Editor.

You can access the latest version of this specification using the link provided.

Using the Streaming API

Creating an event stream

Before you can subscribe to an event stream using the API you will first need to specify the set of events to be streamed using the Mambu UI. We strongly recommend reading our support article on the Streaming API for a walkthrough. Using the UI you will also create the body of each event notification using JSON, XML, or plain text and can make use of placeholders to include specific information such as client or account IDs, transactions amounts, or the values of custom fields.

Once the event streams have been created using the UI you will have a set of topics that you can subscribe to using the POST /subscriptions API by passing them in the event_types array.

Creating Subscriptions

POST /subscriptions

Request

Sample Request

curl -v -X POST "https://TENANT_NAME.mambu.com/api/v1/subscriptions" -H "Content-type: application/json" -d

Body Parameter

{
    "owning_application": "demo-app",
    "event_types": ["mrn.event.TENANT_NAME.streamingapi.client_approved","mrn.event.TENANT_NAME.streamingapi.client_rejected"]
}

A subscription can be created by sending a POST request to the /api/v1/subscriptions resource.

Response

Sample Response

{
  "owning_application": "demo-app",
  "event_types": [
    "mrn.event.TENANT_NAME.streamingapi.client_approved"
  ],
  "consumer_group": "default",
  "read_from": "end",
  "id": "0691160a-b519-4595-b85c-a400fc73e963",
  "created_at": "2018-11-18T22:27:29.156Z"
}

The response returns the whole subscription object that was created, including the server generated id field:

Deleting a Subscription

DELETE /subscriptions/{subscription_id}

Request

Sample Request

curl -X DELETE https://TENANT_NAME.mambu.com/api/v1/subscriptions/071569bc-89f2-4b52-8277-6ed9614ffbb3

A subscription can be removed by sending a DELETE request to the /subscriptions/{subscription_id} resource.

Response

Sample Response

{
    "title": "Not Found",
    "status": 404,
    "detail": "Subscription with id \"071569bc-89f2-4b52-8277-6ed9614ffbb3\" does not exist"
}

The possible responses for the delete are:

204: The subscription has been successfully deleted.
404: The subscription does not exist, along with this error message a response body is received.

Consuming Events from a Subscription

GET /subscriptions/{subscription-id}/events

Request

Sample Request

curl -v -X GET "https://TENANT_NAME.localhost:8889/api/v1/subscriptions/0691160a-b519-4595-b85c-a400fc73e963/events"

Events can be consumed by sending a GET request to the subscription's event resource /subscriptions/{subscription-id}/events.

Response

Sample Response

HTTP/1.1 200 OK
X-Mambu-StreamId: 93ae5174-b863-4f8f-ba33-d274854d1f3d
Transfer-Encoding: chunked

{ "cursor" : {    "partition": "1",    "offset": "001-0001--1",    "event_type": "mrn.event.demo_tenant.streamingapi.client_approved",    "cursor_token": "1d23d044-daa1-414c-8ec4-a060ba03d77c"  },  "info": {    "debug": "Stream started"  }}
{ "cursor" : {    "partition": "1",    "offset": "001-0001-000000000000000000",    "event_type": "mrn.event.demo_tenant.streamingapi.client_approved",    "cursor_token": "e8f6bc5d-65b7-4466-9d95-507da0e79d3c"  },  "events": [    {      "metadata": {        "occurred_at": "2018-11-18T22:50:12.602Z",        "eid": "fde622d5-f975-4786-8e1d-d328c29761f9",        "event_type": "mrn.event.demo_tenant.streamingapi.client_approved",        "partition": "1",        "received_at": "2018-11-18T22:50:12.786Z",        "flow_id": "jOK94UJFdtKsthWOgOJCwuGC",        "version": "1.0.0"      },      "occurred_at": "2018-11-18T22:50:12.602Z",      "body": "Client was approved notification body",      "content_type": "text/plain; charset=UTF-8",      "template_name": "Demo Template 1",      "category": "DATA"    }  ]}
{ "cursor" : {    "partition": "1",    "offset": "001-0001-000000000000000000",    "event_type": "mrn.event.demo_tenant.streamingapi.client_approved",    "cursor_token": "5fed7de4-86ec-464f-92f0-5d6103869385"  }}

The response is a stream that groups events into JSON batches separated by an endline (\n) character. The output can be seen in the right panel.

Each batch contains the following fields:

cursor: The cursor of the batch which should be used for committing the batch.
events: The array of events of this batch.
info: An optional field that can hold useful information (e.g. the reason why the stream was closed by Mambu).

Please Note:
When a stream is started, the client receives a header named X-Mambu-StreamId, which must be used when committing cursors (see below).

Subscription Cursors

Cursor Structure

{
  "partition": "1",
  "offset": "001-0001-000000000000000000",
  "event_type": "mrn.event.demo_tenant.streamingapi.client_approved",
  "cursor_token": "e8f6bc5d-65b7-4466-9d95-507da0e79d3c"
}

The cursors in the Subscription API have the following structure:

The fields are:

partition: The partition this batch belongs to. A batch can only have one partition.
offset: The offset of this batch. The offset is server defined and opaque to the client - clients should not try to infer or assume a structure.
event_type: Specifies the event-type of the cursor (as in one stream there can be events of different event-types)
cursor_token: The cursor token generated by Mambu.

Committing Cursors

POST /subscriptions/{subscriptionId}/cursors

Request

Sample Request

curl -v -X POST "https://TENANT_NAME.mambu.com/api/v1/subscriptions/0691160a-b519-4595-b85c-a400fc73e963/cursors"\
  -H "X-Mambu-StreamId: 93ae5174-b863-4f8f-ba33-d274854d1f3d" \
  -H "Content-type: application/json" \
  -d

Body Parameter

{
    "items": [
      {
        "partition": "1",
        "offset": "001-0001-000000000000000000",
        "event_type": "mrn.event.demo_tenant.streamingapi.client_approved",
        "cursor_token": "db4b4c27-f880-4406-a382-b057acf432cd"
      }    ]
  }

Cursors can be committed by making a POST request to the subscription's cursor resource /subscriptions/{subscriptionId}/cursors:

Please be aware that the X-Mambu-StreamId header is required when committing a cursor. The value should be the same as the X-Mambu-StreamId header you receive when opening a stream of events. Also, each client can only commit batches that it has received.

Response

The possible successful responses for a commit are:

204: Cursors were successfully committed and offset was increased.
200: Cursors were committed but at least one of the cursors didn’t increase the offset, since it was less than or equal to the already committed one. The response body will also include json with a list of cursors and the results of their commits.

The timeout for commit is 60 seconds. If you open the stream, read data, and don’t commit anything for 60 seconds - the stream connection will be closed from the Mambu side. Please note that if there are no events available to send and you get only empty batches - there is no need to commit; Mambu will close the connection only if there is some uncommitted data and no commits happened for 60 seconds.

If the connection is closed for any reason, the client still has 60 seconds to commit the events it received from the moment the events were sent. After that, the session will be considered closed and it will no longer be possible to commit with that X-Mambu-StreamId. If the commit was not made - then the next time you start reading from a subscription you will get data from the last point of your commit, you will receive the events you haven’t committed again and will need to deduplicate.

Java Sample Client

To help get you started quickly, a sample project (in Java) implementing a Streaming API client is available at https://github.com/mambu-gmbh/Streaming-API-Java-Sample-Client.

Streaming API Reference

Create a Subscription

POST /subscriptions

This endpoint creates a subscription for event_types. Event types must first be specified using the Mambu UI following the instructions in our Event Streaming API article.

The subscription is needed to be able to consume events from event_types in a high level way when Mambu stores the offsets and manages the rebalancing of consuming clients.
The subscription is identified by its key parameters (owning_application, event_types, consumer_group).
If this endpoint is invoked several times with the same key subscription properties in body (order of even_types is not important) - the subscription will be created only once and for all other calls it will just return the subscription that was already created.

Example Request

A Subscription

{
  "owning_application": "demo",
  "event_types": [
    "mrn.event.TENANT_NAME.streamingapi.client_approved"
  ],
  "consumer_group": "read-product-updates",
  "read_from": "end",
  "initial_cursors": [
    {
      "partition": "1",
      "offset": "001-0001-000000000000000000",
      "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved"
    }
  ]
}

Parameters

Name	Type	Description	In
apikey (required)	string	your API key	header
body (required)	Subscription	Subscription is a high level consumption unit. Subscriptions allow applications to easily scale the number of clients by managing consumed event offsets and distributing load between instances. The key properties that identify a subscription are `owning_application`, `event_types` and `consumer_group`. It is not possible to have two different subscriptions with these properties being the same.	body

Example Responses

200 :
A subscription

{
  "id": "0691160a-b519-4595-b85c-a400fc73e963",
  "owning_application": "demo",
  "event_types": [
    "mrn.event.TENANT_NAME.streamingapi.client_approved"
  ],
  "consumer_group": "read-product-updates",
  "created_at": "1996-12-19T16:39:57-08:00",
  "updated_at": "1996-12-19T16:39:57-08:00",
  "read_from": "end",
  "initial_cursors": [
    {
      "partition": "1",
      "offset": "001-0001-000000000000000000",
      "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved"
    }
  ]
}

201 :
A newly created subscription

{
  "id": "0691160a-b519-4595-b85c-a400fc73e963",
  "owning_application": "demo",
  "event_types": [
    "mrn.event.TENANT_NAME.streamingapi.client_approved"
  ],
  "consumer_group": "read-product-updates",
  "created_at": "1996-12-19T16:39:57-08:00",
  "updated_at": "1996-12-19T16:39:57-08:00",
  "read_from": "end",
  "initial_cursors": [
    {
      "partition": "1",
      "offset": "001-0001-000000000000000000",
      "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved"
    }
  ]
}

400 Response

{
  "detail": "Connection to database timed out",
  "instance": "http://example.com",
  "status": 503,
  "title": "Service Unavailable",
  "type": "http://httpstatus.es/503"
}

Responses

Status	Meaning	Description	Schema
`200`	OK	Subscription for such parameters already exists. Returns subscription object that already existed.	Subscription
`201`	Created	Subscription was successfuly created. Returns subscription object that was created.	Subscription
`400`	Bad Request	Bad Request	Problem
`422`	Unprocessable Entity	Unprocessable Entity	Problem

Response Headers

Status	Header	Type	Format	Description
200	`Location`	string	uri	The relative URI for this subscription resource.
201	`Location`	string	uri	The relative URI for the created resource.
201	`Content-Location`	string	uri	If the `Content-Location` header is present and the same as the `Location` header the client can assume it has an up to date representation of the Subscription and a corresponding `GET` request is not needed.

Start Event Stream

GET /subscriptions/{subscription_id}/events

This endpoint starts a new stream for reading events from this subscription. The data will be automatically rebalanced between streams of one subscription.

The minimal consumption unit is a partition, so it is possible to start as many streams as the total number of partitions in event-types of this subscription.
The rebalance currently only operates with the number of partitions so the amount of data in event-- types/partitions is not considered during autorebalance.
The position of the consumption is managed by Mambu. The client is required to commit the cursors it gets in a stream.

Example Request

GET https://TENANT_NAME.mambu.com/api/v1/subscriptions/0691160a-b519-4595-b85c-a400fc73e96/events?stream_limit=10&stream_timeout=600 HTTP/1.1

Accept: application/json
apikey: string
X-Flow-Id: string

Client Rebalancing

If you need more than one client for your subscription to distribute load or increase throughput - you can read the subscription with multiple clients and Mambu will automatically balance the load across them.
Currently, the maximum number of supported clients per subscription is equal to the number of event types in the subscription multiplied by 3.
For example if there are two event types in the subscription, the total number of clients for the subscription is 6. The total of all the partitions within a subscription cannot be more than 100. This gives a maximum of 33 event types per subscription.
Recommendation: to improve throughput, maintain the same number of partitions and consumer clients from the start. This ensures a balanced distribution of the workload, minimises delays and simplifies rebalancing.
The API provides a guarantee of at-least-once delivery, this means that there are cases where duplicate events may be sent if there are errors committing events - a useful technique to detect and handle duplicates is to be idempotent and to check the eid field of event metadata.

Parameters

Name	Type	Description	In
X-Flow-Id	string	The flow id of the request, which is written into the logs and passed to called services. Helpful for operational troubleshooting and log analysis.	header
max_uncommitted_events	string	The maximum number of uncommitted events that Mambu will stream before pausing the stream. When in paused state and commit comes - the stream will resume.	query
batch_limit	integer	Maximum number of `event`s in each chunk (and therefore per partition) of the stream.	query
stream_limit	integer	Maximum number of `event`s in this stream (over all partitions being streamed in this connection).	query
batch_flush_timeout	integer	Maximum time in seconds to wait for the flushing of each chunk (per partition).	query
stream_timeout	integer	Maximum time in seconds a stream will live before connection is closed by the server.	query
stream_keep_alive_limit	integer	Maximum number of empty keep alive batches to get in a row before closing the connection.	query
commit_timeout	string	Maximum amount of seconds that Mambu will be waiting for commit after sending a batch to a client.	query
apikey (required)	string	your API key	header
subscription_id (required)	string(uuid)	Id of subscription.	path

Detailed descriptions

batch_limit: Maximum number of events in each chunk (and therefore per partition) of the stream.

If 0 or undefined, will read 1 message in a single batch.
Recommendation: to achieve better efficiency of message processing, read multiple messages in a single batch. This will decrease the number of calls needed for committing offsets.

stream_limit: Maximum number of events in this stream (over all partitions being streamed in this connection).

If 0 or undefined, will stream batches indefinitely.
Stream initialization will fail if stream_limit is lower than batch_limit.

batch_flush_timeout: Maximum time in seconds to wait for the flushing of each chunk (per partition).

If the amount of buffered Events reaches batch_limit before this batch_flush_timeout is reached, the messages are immediately flushed to the client and batch flush timer is reset.
If 0 or undefined, will assume 30 seconds.
Value is treated as a recommendation. Mambu may flush chunks with a smaller timeout.

stream_timeout: Maximum time in seconds a stream will live before connection is closed by the server.

If 0 or unspecified will stream for 1h ±10min.
If this timeout is reached, any pending messages (in the sense of stream_limit) will be flushed to the client.
Stream initialization will fail if stream_timeout is lower than batch_flush_timeout.
If the stream_timeout is greater than max value (4200 seconds) - Mambu will treat this as not specifying stream_timeout.
Recommendation: extend the streaming timeout for improved performance. It will help in preventing the frequent rebalancing that is resulting in message consumption delays.

stream_keep_alive_limit: Maximum number of empty keep alive batches to get in a row before closing the connection.

If 0 or undefined will send keep alive messages indefinitely.

commit_timeout: Maximum amount of seconds that Mambu will be waiting for commit after sending a batch to a client.

In case commit does not come within this timeout, Mambu will initialize stream termination, no new data will be sent. Partitions from this stream will be assigned to other streams.

Setting commit_timeout to 0 is equal to setting it to the maximum allowed value - 60 seconds.

In case low latency is needed, change the commit_timeout to a lower value (e.g. 5 seconds).

Example Responses

200 :
A batch of events

{
  "cursor": {
    "partition": "1",
    "offset": "001-0001-000000000000000000",
    "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
    "cursor_token": "string"
  },
  "info": {},
  "events": [
    {
      "metadata": {
        "eid": "105a76d8-db49-4144-ace7-e683e8f4ba46",
        "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
        "occurred_at": "1996-12-19T16:39:57-08:00",
        "content_type": "text/plain; charset=UTF-8",
        "category": "string"
      },
      "body": "Client was modified. Activity type: CLIENT_SET_TO_INACTIVE. Date: 27-11-2018.\n",
      "template_name": "Client activity template"
    }
  ]
}

400 Response

{
  "detail": "Connection to database timed out",
  "instance": "http://example.com",
  "status": 503,
  "title": "Service Unavailable",
  "type": "http://httpstatus.es/503"
}

Responses

Status	Meaning	Description	Schema
`200`	OK	Stream started. Stream format is a continuous series of `SubscriptionEventStreamBatch` objects separated by `\n`	Subscription-Event-Stream-Batch
`400`	Bad Request	Bad Request.	Problem
`403`	Forbidden	Access forbidden.	Problem
`404`	Not Found	Subscription not found.	Problem
`409`	Conflict	Conflict. There are several possible reasons for receiving this status code: 1) There are no empty slots for this subscriptions. The amount of consumers for this subscription already equals the maximum value - the total number of partitions in this subscription. 2) Request to reset subscription cursors is still in progress.	Problem

Response Headers

Status	Header	Type	Format	Description
200	`X-Mambu-StreamId`	string	uuid	The id of this stream generated by Mambu. Must be used for committing events that were read by client from this stream.

Commit Cursors

POST /subscriptions/{subscription_id}/cursors

This endpoint commits offsets of the subscription.

If there is uncommited data, and no commits happen for 60 seconds, Mambu will consider the client to be gone, and will close the connection. As long as no events are sent, the client does not need to commit.
If the connection is closed, the client has 60 seconds to commit the events it received, from the moment they were sent. After that, the connection will be considered closed, and it will not be possible to do commit with that X-Mambu-StreamId anymore.
When a batch is committed, that also automatically commits all previous batches that were sent in a stream for this partition.

Example Request

A Cursor

{
  "items": [
    {
      "partition": "1",
      "offset": "001-0001-000000000000000000",
      "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
      "cursor_token": "string"
    }
  ]
}

Parameters

Name	Type	Description	In
X-Mambu-StreamId (required)	string	Id of stream which client uses to read events. It is not possible to make a commit for a terminated or none-existing stream. Also the client can't commit something which was not sent to his stream.	header
apikey (required)	string	your API key	header
body	object	none	body
» items (required)	[object]	List of cursors that the consumer acknowledges to have successfully processed.	body
»» partition (required)	string	Id of the partition pointed to by this cursor.	body
»» offset (required)	string	Offset of the event being pointed to. Note that if you want to specify beginning position of a stream with first event at offset `N`, you should specify offset `N-1`.	body
»» event_type (required)	string	The name of the event type this partition's events belong to.	body
»» cursor_token (required)	string	An opaque value defined by the server.	body
subscription_id (required)	string	Id of subscription	path

Detailed descriptions

»» offset: Offset of the event being pointed to. Note that if you want to specify beginning position of a stream with first event at offset N, you should specify offset N-1.

This applies in cases when you create new subscription or reset subscription offsets.

Also for stream start offsets one can use two special values:

begin - read from the oldest available event.
end - read from the most recent offset.

Example Responses

200 :
A Cursor

{
  "items": [
    {
      "cursor": {
        "partition": "1",
        "offset": "001-0001-000000000000000000",
        "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
        "cursor_token": "string"
      },
      "result": "string"
    }
  ]
}

403 Response

{
  "detail": "Connection to database timed out",
  "instance": "http://example.com",
  "status": 503,
  "title": "Service Unavailable",
  "type": "http://httpstatus.es/503"
}

Responses

Status	Meaning	Description	Schema
`200`	OK	At least one cursor which was tried to be committed is older or equal to already committed one. Array of commit results is returned for this status code	Inline
`204`	No Content	Offsets were committed	None
`403`	Forbidden	Access forbidden	Problem
`404`	Not Found	Subscription not found	Problem
`422`	Unprocessable Entity	Unprocessable Entity	Problem

Response Schema

Status Code 200

Name	Type	Description
» items (required)	[Cursor-Commit-Result]	[The result of single cursor commit. Holds a cursor itself and a result value.]
»» cursor (required)	any	none

allOf

Name	Type	Description
»»» anonymous	object	none
»»»» cursor_token (required)	string	An opaque value defined by the server.
»»»» event_type (required)	string	The name of the event type this partition's events belong to.

and

Name	Type	Description
»»» anonymous	Cursor	none
»»»» offset (required)	string	Offset of the event being pointed to. Note that if you want to specify beginning position of a stream with first event at offset `N`, you should specify offset `N-1`. This applies in cases when you create new subscription or reset subscription offsets. Also for stream start offsets one can use two special values: - `begin` - read from the oldest available event. - `end` - read from the most recent offset.
»»»» partition (required)	string	Id of the partition pointed to by this cursor.

continued

Name	Type	Description
»» result (required)	string	The result of cursor commit. - `committed`: cursor was successfully committed - `outdated`: there already was more recent (or the same) cursor committed, so the current one was not committed as it is outdated

Delete a Subscription

DELETE /subscriptions/{subscription_id}

This endpoint deletes an existing subscription for event types.

Example Request

DELETE https://TENNT_NAME.MAMBU.COM/api/v1/subscriptions/0691160a-b519-4595-b85c-a400fc73e96 HTTP/1.1

apikey: string

In case the subscription is not needed anymore, it can be manually deleted by providing its unique subscription id.

Parameters

Name	Type	Description	In
apikey (required)	string	your API key	header
subscription_id (required)	string	ID of subscription	path

Example Responses

404 :
Subscription ID not found

{
  "title": "Not Found",
  "status": 404,
  "detail": "Subscription with id \"071569bc-89f2-4b52-8277-6ed9614ffbb3\" does not exist"
}

Responses

Status	Meaning	Description	Schema
`204`	No Content	Subscription for the given id has been successfully deleted	None
`404`	Not Found	Subscription for the given id was not found	Inline

Response Schema

Status Code 404

Name	Type	Description
» title (required)	string	A short, summary of the problem type. Written in English and readable for engineers (usually not suited for non technical stakeholders and not localized).
» status (required)	number	The HTTP status code generated by the origin server for this occurrence of the problem.
» detail (required)	string	A human readable explanation specific to this occurrence of the problem.

Get Statistics

GET /subscriptions/{subscription_id}/stats

This endpoint exposes statistics of a specified subscription. An example use for the statistics provided could be monitoring lag in consumers.

Example Request

GET https://TENANT_NAME.mambu.com/api/v1/subscriptions/0691160a-b519-4595-b85c-a400fc73e96/stats?show_time_lag=true HTTP/1.1

Accept: application/json
apikey: string

The latest offset is compared with committed offset in order to calculate unconsumed events count for specific partition.

Parameters

Name	Type	Description	In
show_time_lag	boolean	Shows consumer time lag as an optional field if requested. This option is a time consuming operation and Streaming API attempts to compute it in the best possible strategy. In cases of failures, resulting in Streaming API being unable to compute it within a configurable timeout, the field might either be partially present or not present (depending on the number of successful requests) in the output.	query
apikey (required)	string	your API key	header
subscription_id (required)	string	The ID of the subscription	path

Example Responses

200 :
example statistics

{
  "items": [
    {
      "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
      "partitions": [
        {
          "partition": "0",
          "state": "unassigned",
          "unconsumed_events": 1,
          "consumer_lag_seconds": 500,
          "stream_id": ""
        },
        {
          "partition": "1",
          "state": "unassigned",
          "unconsumed_events": 0,
          "consumer_lag_seconds": 0,
          "stream_id": ""
        },
        {
          "partition": "2",
          "state": "unassigned",
          "unconsumed_events": 0,
          "consumer_lag_seconds": 0,
          "stream_id": ""
        }
      ]
    }
  ]
}

404 Response

{
  "detail": "Connection to database timed out",
  "instance": "http://example.com",
  "status": 503,
  "title": "Service Unavailable",
  "type": "http://httpstatus.es/503"
}

Responses

Status	Meaning	Description	Schema
`200`	OK	Statistics list for specified subscription successfully returned.	Inline
`404`	Not Found	Subscription for the given id was not found.	Problem

Response Schema

Status Code 200

Name	Type	Description
» items (required)	[SubscriptionEventTypeStats]	Statistics list for specified subscription.
»» SubscriptionEventTypeStats	SubscriptionEventTypeStats	Statistics of one event-type within a context of subscription.
»»» event_type (required)	string	The name of the event type this partition's events belong to.
»»» partitions (required)	[PartitionStats]	Statistics of partitions of this event-type.
»»»» consumer_lag_seconds	integer	Subscription consumer lag for this partition in seconds. Measured as the age of the oldest event of this partition that is not yet consumed within this subscription.
»»»» partition (required)	string	The partition id.
»»»» state (required)	string	The state of this partition in current subscription. Currently following values are possible: `unassigned` - the partition is currently not assigned to any client. `reassigning` - the partition is currently reassigning from one client to another. `assigned` - the partition is assigned to a client.
»»»» stream_id (required)	string	The id of the stream that consumes data from this partition.
»»»» unconsumed_events (required)	integer	The amount of events in this partition that are not yet consumed within this subscription. The property may be absent at the moment when no events were yet consumed from the partition in this subscription (In case of read_from is `BEGIN` or `END`). If the event type uses ‘compact’ cleanup policy - then the actual number of unconsumed events can be lower than the one reported in this field.

Enumerated Values

Property	Value
state	`assigned`
state	`unassigned`
state	`reassigning`

Schemas

Cursor

{
  "offset": "001-0001-000000000000000000",
  "partition": "1"
}

Properties

Name	Type	Description	Restrictions
offset (required)	string	Offset of the event being pointed to. Note that if you want to specify beginning position of a stream with first event at offset `N`, you should specify offset `N-1`. This applies in cases when you create new subscription or reset subscription offsets. Also for stream start offsets one can use two special values: - `begin` - read from the oldest available event. - `end` - read from the most recent offset.	none
partition (required)	string	Id of the partition pointed to by this cursor.	none

Cursor-Commit-Result

{
  "cursor": {
    "cursor_token": "string",
    "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
    "offset": "001-0001-000000000000000000",
    "partition": "1"
  },
  "result": "string"
}

The result of single cursor commit. Holds a cursor itself and a result value.

Properties

Name	Type	Description	Restrictions
cursor (required)	Subscription-Cursor	none	none
result (required)	string	The result of cursor commit. - `committed`: cursor was successfully committed - `outdated`: there already was more recent (or the same) cursor committed, so the current one was not committed as it is outdated	none

Event

{
  "body": "Client was modified. Activity type: CLIENT_SET_TO_INACTIVE. Date: 27-11-2018.\n",
  "metadata": {
    "category": "string",
    "content_type": "text/plain; charset=UTF-8",
    "eid": "105a76d8-db49-4144-ace7-e683e8f4ba46",
    "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
    "occurred_at": "1996-12-19T16:39:57-08:00"
  },
  "template_name": "Client activity template"
}

Payload of an Event. Usually represents a status transition in a Business process.

Properties

Name	Type	Description	Restrictions
body (required)	string	Actual content of the notification.	none
metadata (required)	Event-Metadata	Metadata for this Event.	none
template_name (required)	string	Name of the notification template.	none

Event-Metadata

{
  "category": "string",
  "content_type": "text/plain; charset=UTF-8",
  "eid": "105a76d8-db49-4144-ace7-e683e8f4ba46",
  "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
  "occurred_at": "1996-12-19T16:39:57-08:00"
}

Metadata for this Event.

Properties

Name	Type	Description	Restrictions
category (required)	string	Indicates if the content of the notification can be configured in Mambu or it is fixed. Currently only one category is supported: `DATA`	none
content_type (required)	string	Notification content format.	none
eid (required)	string(uuid)	Unique identifier of this Event. Consumers MIGHT use this value to assert uniqueness of reception of the Event.	none
event_type (required)	string	The EventType of this Event.	none
occurred_at (required)	string(date-time)	Timestamp of creation of the Event generated by Mambu.	none

Enumerated Values

Property	Value
content_type	`application/xml`
content_type	`application/json`
content_type	`text/plain; charset=UTF-8`

PartitionStats

{
  "consumer_lag_seconds": 0,
  "partition": "0",
  "state": "unassigned",
  "stream_id": "",
  "unconsumed_events": 0
}

Statistics of partition within a subscription context.

Properties

Name	Type	Description	Restrictions
consumer_lag_seconds	integer	Subscription consumer lag for this partition in seconds. Measured as the age of the oldest event of this partition that is not yet consumed within this subscription.	none
partition (required)	string	The partition id.	none
state (required)	string	The state of this partition in current subscription. Currently following values are possible: `unassigned` - the partition is currently not assigned to any client. `reassigning` - the partition is currently reassigning from one client to another. `assigned` - the partition is assigned to a client.	none
stream_id (required)	string	The id of the stream that consumes data from this partition.	none
unconsumed_events (required)	integer	The amount of events in this partition that are not yet consumed within this subscription. The property may be absent at the moment when no events were yet consumed from the partition in this subscription (In case of read_from is `BEGIN` or `END`). If the event type uses ‘compact’ cleanup policy - then the actual number of unconsumed events can be lower than the one reported in this field.	none

Enumerated Values

Property	Value
state	`assigned`
state	`unassigned`
state	`reassigning`

Problem

{
  "detail": "Connection to database timed out",
  "instance": "http://example.com",
  "status": 503,
  "title": "Service Unavailable",
  "type": "http://httpstatus.es/503"
}

Properties

Name	Type	Description	Restrictions
detail	string	A human readable explanation specific to this occurrence of the problem.	none
instance	string(uri)	An absolute URI that identifies the specific occurrence of the problem. It may or may not yield further information if dereferenced.	none
status (required)	integer	The HTTP status code generated by the origin server for this occurrence of the problem.	none
title (required)	string	A short, summary of the problem type. Written in English and readable for engineers (usually not suited for non technical stakeholders and not localized).	none
type (required)	string(uri)	An absolute URI that identifies the problem type. When dereferenced, it SHOULD provide human-readable API documentation for the problem type (e.g., using HTML).	none

Stream-Info

{}

This object contains general information about the stream. Used only for debugging purposes. We recommend logging this object in order to solve connection issues.

Clients should not parse this structure.

Properties

None

Subscription

{
  "consumer_group": "read-product-updates",
  "created_at": "1996-12-19T16:39:57-08:00",
  "event_types": [
    "mrn.event.TENANT_NAME.streamingapi.client_approved"
  ],
  "id": "0691160a-b519-4595-b85c-a400fc73e963",
  "initial_cursors": [
    {
      "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
      "offset": "001-0001-000000000000000000",
      "partition": "1"
    }
  ],
  "owning_application": "demo",
  "read_from": "end",
  "updated_at": "1996-12-19T16:39:57-08:00"
}

Subscription is a high level consumption unit. Subscriptions allow applications to easily scale the number of clients by managing consumed event offsets and distributing load between instances. The key properties that identify a subscription are owning_application, event_types and consumer_group. It is not possible to have two different subscriptions with these properties being the same.

Properties

Name	Type	Description	Restrictions
consumer_group	string	The value describing the use case of this subscription. In general that is an additional identifier used to differ subscriptions having the same `owning_application` and `event_types`.	none
created_at	string(date-time)	Timestamp of creation of the subscription. This is generated by Mambu. It should not be specified when creating subscription and sending it may result in a client error.	read-only
event_types (required)	[string]	EventTypes to subscribe to. The order is not important. Subscriptions that differ only by the order of EventTypes will be considered the same and will have the same id. The size of the `event_type`s list is limited by the total number of partitions within these event types. Default limit for partition count is `100`.	none
id	string	ID of subscription that was created. Is generated by Mambu, should not be specified when creating a subscription.	none
initial_cursors	[Subscription-Cursor-Without-Token]	List of cursors to start reading from. This property is required when `read_from` = cursors. The `initial` cursors should cover all partitions of subscription. Clients will get events starting from next offset positions.	none
owning_application (required)	string	The id of application owning the subscription.	none
read_from	string	Position to start reading events from. Currently supported values: - `begin` - read from the oldest available event. - `end` - read from the most recent offset. - `cursors` - read from cursors provided in initial_cursors property. Applied when the client starts reading from a subscription.	none
updated_at	string(date-time)	Timestamp of last update of the subscription. This is generated by Mambu. It should not be specified when creating subscription and sending it may result in a client error. Its initial value is same as `created_at`.	read-only

Subscription-Cursor

{
  "cursor_token": "string",
  "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
  "offset": "001-0001-000000000000000000",
  "partition": "1"
}

Properties

allOf

Name	Type	Description	Restrictions
*anonymous*	object	none	none
» cursor_token (required)	string	An opaque value defined by the server.	none
» event_type (required)	string	The name of the event type this partition's events belong to.	none

and

Name	Type	Description	Restrictions
*anonymous*	Cursor	none	none

Subscription-Cursor-Without-Token

{
  "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
  "offset": "001-0001-000000000000000000",
  "partition": "1"
}

Properties

allOf

Name	Type	Description	Restrictions
*anonymous*	Cursor	none	none

and

Name	Type	Description	Restrictions
*anonymous*	object	none	none
» event_type (required)	string	The name of the event type this partition's events belong to.	none

Subscription-Event-Stream-Batch

{
  "cursor": {
    "cursor_token": "string",
    "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
    "offset": "001-0001-000000000000000000",
    "partition": "1"
  },
  "events": [
    {
      "body": "Client was modified. Activity type: CLIENT_SET_TO_INACTIVE. Date: 27-11-2018.\n",
      "metadata": {
        "category": "string",
        "content_type": "text/plain; charset=UTF-8",
        "eid": "105a76d8-db49-4144-ace7-e683e8f4ba46",
        "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
        "occurred_at": "1996-12-19T16:39:57-08:00"
      },
      "template_name": "Client activity template"
    }
  ],
  "info": {}
}

One chunk of events in a stream. A batch consists of an array of events plus a cursor pointing to the offset of the last event in the stream.

The size of the array of event is limited by the parameters used to initialize a Stream.

Sequential batches might present repeated cursors if no new events have arrived.

Properties

Name	Type	Description	Restrictions
cursor (required)	Subscription-Cursor	none	none
events	[Event]	[Payload of an Event. Usually represents a status transition in a Business process.]	none
info	Stream-Info	This object contains general information about the stream. Used only for debugging purposes. We recommend logging this object in order to solve connection issues. Clients should not parse this structure.	none

SubscriptionEventTypeStats

{
  "event_type": "mrn.event.TENANT_NAME.streamingapi.client_approved",
  "partitions": [
    {
      "consumer_lag_seconds": 500,
      "partition": "0",
      "state": "unassigned",
      "stream_id": "",
      "unconsumed_events": 1
    },
    {
      "consumer_lag_seconds": 0,
      "partition": "1",
      "state": "unassigned",
      "stream_id": "",
      "unconsumed_events": 0
    },
    {
      "consumer_lag_seconds": 0,
      "partition": "2",
      "state": "unassigned",
      "stream_id": "",
      "unconsumed_events": 0
    }
  ]
}

SubscriptionEventTypeStats

Properties

Name	Type	Description	Restrictions
event_type (required)	string	The name of the event type this partition's events belong to.	none
partitions (required)	[PartitionStats]	Statistics of partitions of this event-type.	none