Event Handling

The state layer provides a way to be notified about state updates easily. However, it’s also possible to watch for WebSocket events directly by collecting an events flow, should you need it.

Observing events

The FeedsClient exposes an events property to get a flow of events:

feedsClient
    .events
    .collect { event ->
        // Handle the event types you care about
        when (event) {
            is ActivityAddedEvent -> {
                println("New activity added: ${event.activity.text}")
            }
            is CommentAddedEvent -> {
                println("New comment added: ${event.comment.text}")
            }
            is FollowCreatedEvent -> {
                println("New follow created: ${event.follow.targetFeed}")
            }
            else -> {
                println("Received event: ${event::class.java.simpleName}")
            }
        }
    }

Leveraging the Flow API

With the events coming as a Flow, no special knowledge is needed to get started. You can use all the power of Kotlin's coroutines and Flow API to handle events in a way that suits your use cases.

For example, if you want to only listen for a specific event type, like new activities added, you can use the filterIsInstance operator:

feedsClient
    .events
    .filterIsInstance<ActivityAddedEvent>()
    .collect { event ->
        println("New activity added: ${event.activity.text}")
    }

Handling backpressure

If your event processing is slow, you might want to add explicit backpressure handling because the events flow will drop events if the consumer cannot keep up.

For example, you can use the buffer operator and specify your desired strategy:

feedsClient
    .events
    .buffer(capacity = 100, onBufferOverflow = BufferOverflow.DROP_OLDEST)
    .collect { event ->
        // Process the event
    }

Available Event Types

Below is a table of all available events:

You can configure your Stream app to receive webhook events as well as AWS SNS and AWS SQS. Webhooks are usually the simplest way to receive events from your app and to perform additional action based on what happens to your application.

The configuration can be done using the API or from the Dashboard. By default, all events are sent to your webhook/sqs/sns endpoint, you can also configure the events you want to receive in the dashboard.

await serverAuthClient.updateAppSettings({
  event_hooks: [
    {
      hook_type: "webhook",
      enabled: true,
      // Pass an empty array to subscribe to all events
      event_types: [],
      webhook_url: "<webhook url>",
    },
    {
      hook_type: "webhook",
      enabled: true,
      // Subscribe only to specific events
      event_types: ["feeds.activity.added" /* ... */],
      webhook_url: "<webhook url>",
    },
  ],
});

client.update_app(
    event_hooks=[
        {
            "hook_type": "webhook",
            "enabled": True,
            # Pass an empty list to subscribe to all events
            "event_types": [],
            "webhook_url": "<webhook url>",
        },
        {
            "hook_type": "webhook",
            "enabled": True,
            # Subscribe only to specific events
            "event_types": ["feeds.activity.added"],
            "webhook_url": "<webhook url>",
        },
    ]
)

request = GetStream::Generated::Models::UpdateAppRequest.new(
  event_hooks: [
    {
      hook_type: "webhook",
      enabled: true,
      # Pass an empty array to subscribe to all events
      event_types: [],
      webhook_url: "<webhook url>",
    },
    {
      hook_type: "webhook",
      enabled: true,
      # Subscribe only to specific events
      event_types: ["feeds.activity.added"],
      webhook_url: "<webhook url>",
    },
  ]
)

client.common.update_app(request)

use GetStream\GeneratedModels\UpdateAppRequest;
use GetStream\GeneratedModels\EventHook;

$request = new UpdateAppRequest(
    eventHooks: [
        new EventHook(
            hookType: 'webhook',
            enabled: true,
            eventTypes: [], // empty array = all events
            webhookUrl: '<webhook url>',
        ),
        new EventHook(
            hookType: 'webhook',
            enabled: true,
            eventTypes: ['feeds.activity.added'], // specific events
            webhookUrl: '<webhook url>',
        ),
    ]
);

$client->updateApp($request);

_, err = client.UpdateApp(ctx, &getstream.UpdateAppRequest{
  EventHooks: []getstream.EventHook{
    {
      HookType:    getstream.PtrTo("webhook"),
      Enabled:     getstream.PtrTo(true),
      // Pass an empty array to subscribe to all events
      EventTypes:  []string{},
      WebhookUrl:  getstream.PtrTo("<webhook url>"),
    },
    {
      HookType:    getstream.PtrTo("webhook"),
      Enabled:     getstream.PtrTo(true),
      // Subscribe only to specific events
      EventTypes:  []string{"feeds.activity.added"},
      WebhookUrl:  getstream.PtrTo("<webhook url>"),
    },
  },
})
if err != nil {
  log.Fatal("Error updating app settings:", err)
}

var webhookHooks = new List<EventHook>
{
  new EventHook
  {
    HookType = HookType.Webhook,
    Enabled = true,
    EventTypes = new List<string>(), // empty list = all events
    WebhookUrl = "<webhook url>",
  },
  new EventHook
  {
    HookType = HookType.Webhook,
    Enabled = true,
    EventTypes = new List<string> { "feeds.activity.added" }, // specific events
    WebhookUrl = "<webhook url>",
  },
};

await client.UpdateAppAsync(new UpdateAppRequest
{
  EventHooks = webhookHooks,
});

import io.getstream.models.EventHook;
import io.getstream.models.UpdateAppRequest;

EventHook allEventsHook = new EventHook()
    .hookType(EventHook.HookTypeEnum.WEBHOOK)
    .enabled(true)
    .eventTypes(Collections.emptyList()) // empty list = all events
    .webhookUrl("<webhook url>");

EventHook specificEventsHook = new EventHook()
    .hookType(EventHook.HookTypeEnum.WEBHOOK)
    .enabled(true)
    .eventTypes(List.of("feeds.activity.added")) // specific events
    .webhookUrl("<webhook url>");

UpdateAppRequest request = new UpdateAppRequest()
    .eventHooks(List.of(allEventsHook, specificEventsHook));

common.updateApp(request).execute();

Some important points to consider:

• The selection of events you want to receive applies to all the endpoints you have configured.
• You can configure multiple endpoints for the same app (eg. AWS SNS and HTTP Webhook).
• If your app is configured to receive all events, you can still filter the events you want to receive in your webhook handler.
• If your app is configured to receive all events, newly introduced event types will be sent to your webhook handler by default.
• If you pick specific events, newly introduced event types will not be sent to your webhook handler by default (you can still manually add them later on).

How to implement a webhook handler

Your webhook handler needs to follow these rules:

• accept HTTP POST requests with JSON payload
• be reachable from the public internet. Tunneling services like Ngrok are supported
• respond with response codes from 200 to 299 as fast as possible

Your webhook handler can use the type field to handle events based correctly based on their type and payload.

All webhook requests contain these headers:

Best Practices

We highly recommend following common security guidelines to make your webhook integration safe and fast:

• Use HTTPS with a certificate from a trusted authority
• Verify the "X-Signature" header to ensure the request is coming from Stream
• Support HTTP Keep-Alive
• Use a highly available infrastructure such as AWS Elastic Load Balancer, Google Cloud Load Balancer, or similar
• Offload the processing of the message if possible (read, store, and forget)
• When decoding JSON into objects, ensure that your webhook can handle new fields being added to the JSON payload as well as new event types (eg. log unknown fields and event types instead of failing)

Error Handling

In case of the request failure Stream Chat attempts to retry a request. The amount of maximum attempts depends on the kind of the error it receives:

• Response code is 408, 429 or >=500: 3 attempts
• Network error: 2 attempts
• Request timeout: 3 attempts

The timeout of one request is 6 seconds, and the request with all retries cannot exceed the duration of 15 seconds.

Available Event Types

Below is a comprehensive table of all available event types and their descriptions. WebhookEvent model definition can be found in Open API specification.