Persistent subscriptions

This document explains how to use the HTTP API for setting up and consuming persistent subscriptions and competing consumer subscription groups. For an overview on competing consumers and how they relate to other subscription types, please see our getting started guide.

TIP

The Administration UI includes a Competing Consumers section where you are able to create, update, delete and view subscriptions and their statuses.

Creating a persistent subscription

Before interacting with a subscription group, you need to create one. This requires admin permissions. You will get an error if you try to create a subscription group more than once.

WARNING

Persistent subscriptions to $all are not supported over the HTTP API. If you want to create persistent subscriptions to $all, use the appropriate client method.

URISupported Content TypesMethod
/subscriptions/{stream}/{subscription_name}application/jsonPUT

Query parameters

ParameterDescription
streamThe stream the persistent subscription is on.
subscription_nameThe name of the subscription group.

Body

ParameterDescription
resolveLinktosTells the subscription to resolve link events.
startFromStart the subscription from the position of the event in the stream.
extraStatisticsTells the backend to measure timings on the clients so statistics will contain histograms of them.
checkPointAfterMillisecondsThe amount of time the system should wait before trying to checkpoint.
liveBufferSizeThe size of the live buffer (in memory) before resorting to paging.
readBatchSizeThe size of the read batch when in paging mode.
bufferSizeThe number of messages that should be buffered when in paging mode.
maxCheckPointCountThe maximum number of messages not checkpointed before forcing a checkpoint.
maxRetryCountSets the number of times a message should be retried before it is considered a bad message.
maxSubscriberCountSets the maximum number of allowed TCP subscribers.
messageTimeoutMillisecondsSets the timeout for a client before the message will be retried.
minCheckPointCountThe minimum number of messages to write a checkpoint for.
namedConsumerStrategyAllowed values are RoundRobin, DispatchToSingle, Pinned, PinnedByCorrelation.

Updating a persistent subscription

You can edit the settings of an existing subscription while it is running. This drops the current subscribers and resets the subscription internally. This requires admin permissions.

URISupported Content TypesMethod
/subscriptions/{stream}/{subscription_name}application/jsonPOST

WARNING

Persistent subscriptions to $all are not supported over the HTTP API. To update persistent subscriptions to $all, use the appropriate client method.

Query parameters

ParameterDescription
streamThe stream the persistent subscription is on.
subscription_nameThe name of the subscription group.

Body

Same parameters as Creating a Persistent Subscription.

Deleting a persistent subscription

URISupported Content TypesMethod
/subscriptions/{stream}/{subscription_name}application/jsonDELETE

WARNING

Deleting persistent subscriptions to $all is not supported over the HTTP API. If you want to delete persistent subscriptions to $all, use the appropriate client method.

Query parameters

ParameterDescription
streamThe stream the persistent subscription is on.
subscription_nameThe name of the subscription group.

Reading a stream via a persistent subscription

By default, reading a stream via a persistent subscription returns a single event per request and does not embed the event properties as part of the response.

URISupported Content TypesMethod
/subscriptions/{stream}/{subscription_name} /subscriptions/{stream}/{subscription_name}?embed={embed} /subscriptions/{stream}/{subscription}/{count}?embed={embed}application/vnd.eventstore.competingatom+xml application/vnd.eventstore.competingatom+jsonGET

Query parameters

ParameterDescription
streamThe stream the persistent subscription is on.
subscription_nameThe name of the subscription group.
countHow many events to return for the request.
embedAllowed values are None, Content, Rich, Body, PrettyBody, TryHarder.

See Reading Streams for information on the different embed levels.

Response

{
  "title": "All Events Persistent Subscription",
  "id": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1",
  "updated": "2015-12-02T09:17:48.556545Z",
  "author": {
    "name": "EventStore"
  },
  "headOfStream": false,
  "links": [
    {
      "uri": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/ack%3Fids=c322e299-cb73-4b47-97c5-5054f920746f",
      "relation": "ackAll"
    },
    {
      "uri": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/nack%3Fids=c322e299-cb73-4b47-97c5-5054f920746f",
      "relation": "nackAll"
    },
    {
      "uri": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/1%3Fembed=None",
      "relation": "previous"
    },
    {
      "uri": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1",
      "relation": "self"
    }
  ],
  "entries": [
    {
      "title": "1@newstream",
      "id": "http://localhost:2113/streams/newstream/1",
      "updated": "2015-12-02T09:17:48.556545Z",
      "author": {
        "name": "EventStore"
      },
      "summary": "SomeEvent",
      "links": [
        {
          "uri": "http://localhost:2113/streams/newstream/1",
          "relation": "edit"
        },
        {
          "uri": "http://localhost:2113/streams/newstream/1",
          "relation": "alternate"
        },
        {
          "uri": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/ack/c322e299-cb73-4b47-97c5-5054f920746f",
          "relation": "ack"
        },
        {
          "uri": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/nack/c322e299-cb73-4b47-97c5-5054f920746f",
          "relation": "nack"
        }
      ]
    }
  ]
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

Acknowledgements

Clients must acknowledge (or not acknowledge) messages in the competing consumer model. If processing is successful, send an ack (acknowledge) to the server to let it know that the message has been handled. If processing fails, then you can nack (not acknowledge) the message and tell the server how to handle the failure. If the client fails to respond in the given timeout period, the message is retried. You should use the rel links in the feed for acknowledgements rather than bookmark URIs, as they are subject to change in future versions.

For example:

{
  "uri": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/ack/c322e299-cb73-4b47-97c5-5054f920746f",
  "relation": "ack"
}
1
2
3
4

Ack multiple messages

URISupported Content TypesMethod
/subscriptions/{stream}/{subscription_name}/ack?ids={messageids}application/jsonPOST

Query parameters

ParameterDescription
streamThe stream the persistent subscription is on.
subscription_nameThe name of the subscription group.
messageidsThe IDs of the messages that needs to be acknowledged.

Ack a single message

URISupported Content TypesMethod
/subscriptions/{stream}/{subscription_name}/ack/{messageid}application/jsonPOST

Query parameters

ParameterDescription
streamThe stream to the persistent subscription is on.
subscription_nameThe name of the subscription group.
messageidThe ID of the message that needs to be acknowledged.

Nack multiple messages

URISupported Content TypesMethod
/subscriptions/{stream}/{subscription_name}/nack?ids={messageids}?action={action}application/jsonPOST

Query parameters

ParameterDescription
streamThe stream to the persistent subscription is on.
subscription_nameThe name of the subscription group.
action
  • Park: Don't retry the message; park it until a request is sent to replay the parked messages.
  • Retry: Retry the message.
  • Skip: Discard the message.
  • Stop: Stop the subscription.
messageidThe ID of the message that needs to be nacked.

Nack a single message

URISupported Content TypesMethod
/subscriptions/{stream}/{subscription_name}/nack/{messageid}?action={action}application/jsonPOST

Replaying parked messages

URISupported Content TypesMethod
/subscriptions/{stream}/{subscription_name}/replayParkedapplication/jsonPOST

Getting information for all subscriptions

URIMethod
/subscriptionsGET

Response

[
  {
    "links": [
      {
        "href": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/info",
        "rel": "detail"
      }
    ],
    "eventStreamId": "newstream",
    "groupName": "competing_consumers_group1",
    "parkedMessageUri": "http://localhost:2113/streams/$persistentsubscription-newstream::competing_consumers_group1-parked",
    "getMessagesUri": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/1",
    "status": "Live",
    "averageItemsPerSecond": 0.0,
    "totalItemsProcessed": 0,
    "lastProcessedEventNumber": -1,
    "lastKnownEventNumber": 5,
    "connectionCount": 0,
    "totalInFlightMessages": 0
  },
  {
    "links": [
      {
        "href": "http://localhost:2113/subscriptions/another_newstream/competing_consumers_group1/info",
        "rel": "detail"
      }
    ],
    "eventStreamId": "another_newstream",
    "groupName": "competing_consumers_group1",
    "parkedMessageUri": "http://localhost:2113/streams/$persistentsubscription-another_newstream::competing_consumers_group1-parked",
    "getMessagesUri": "http://localhost:2113/subscriptions/another_newstream/competing_consumers_group1/1",
    "status": "Live",
    "averageItemsPerSecond": 0.0,
    "totalItemsProcessed": 0,
    "lastProcessedEventNumber": -1,
    "lastKnownEventNumber": -1,
    "connectionCount": 0,
    "totalInFlightMessages": 0
  }
]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

Get subscriptions for a stream

URISupported Content TypesMethod
/subscriptions/{stream}application/jsonGET

Response

[
  {
    "links": [
      {
        "href": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/info",
        "rel": "detail"
      }
    ],
    "eventStreamId": "newstream",
    "groupName": "competing_consumers_group1",
    "parkedMessageUri": "http://localhost:2113/streams/$persistentsubscription-newstream::competing_consumers_group1-parked",
    "getMessagesUri": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/1",
    "status": "Live",
    "averageItemsPerSecond": 0.0,
    "totalItemsProcessed": 0,
    "lastProcessedEventNumber": -1,
    "lastKnownEventNumber": 5,
    "connectionCount": 0,
    "totalInFlightMessages": 0
  }
]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

Getting a specific subscription

URISupported Content TypesMethod
/subscriptions/{stream}/{subscription_name}/infoapplication/jsonGET

Response

{
  "links": [
    {
      "href": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/info",
      "rel": "detail"
    },
    {
      "href": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/replayParked",
      "rel": "replayParked"
    }
  ],
  "config": {
    "resolveLinktos": false,
    "startFrom": 0,
    "messageTimeoutMilliseconds": 10000,
    "extraStatistics": false,
    "maxRetryCount": 10,
    "liveBufferSize": 500,
    "bufferSize": 500,
    "readBatchSize": 20,
    "preferRoundRobin": true,
    "checkPointAfterMilliseconds": 1000,
    "minCheckPointCount": 10,
    "maxCheckPointCount": 500,
    "maxSubscriberCount": 10,
    "namedConsumerStrategy": "RoundRobin"
  },
  "eventStreamId": "newstream",
  "groupName": "competing_consumers_group1",
  "status": "Live",
  "averageItemsPerSecond": 0.0,
  "parkedMessageUri": "http://localhost:2113/streams/$persistentsubscription-newstream::competing_consumers_group1-parked",
  "getMessagesUri": "http://localhost:2113/subscriptions/newstream/competing_consumers_group1/1",
  "totalItemsProcessed": 0,
  "countSinceLastMeasurement": 0,
  "lastProcessedEventNumber": -1,
  "lastKnownEventNumber": 5,
  "readBufferCount": 6,
  "liveBufferCount": 5,
  "retryBufferCount": 0,
  "totalInFlightMessages": 0,
  "connections": []
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
Last Updated:
Contributors: Claude Devarenne