Tyk Streams – Manage Event-Driven APIs

​

Overview

​

Why use Tyk Streams

• Protocol Mediation: Tyk Streams can mediate between different asynchronous protocols and API styles, such as WebSocket, Server-Sent Events (SSE), and Webhooks. This allows you to expose your event streams in a format compatible with your consumers’ requirements.
• Security: Apply the same security policies and controls to your async APIs as you do to your synchronous APIs. This includes features like authentication and authorization.
• Transformations: Transform and enrich your event data on the fly using Tyk’s powerful middleware and plugin system. This allows you to adapt your event streams to meet the needs of different consumers.
• Analytics: Monitor the usage and performance of your async APIs with detailed analytics and reporting. Gain insights into consumer behavior and system health.
• Developer Portal: Publish your async APIs to the Tyk Developer Portal, which provides a centralised catalog for discovery, documentation, and subscription management.

​

Getting Started

​

Example Scenario

​

Tyk as an API Producer Gateway (Client to Stream)

• Goal: Allow a client (like a browser or a script) to easily send a message into our asynchronous system without needing direct access or knowledge of the backend message broker (Kafka in this case).
• Scenario: The user types “Tell me a joke” into the chat interface and hits send.
• Flow:
  1. The browser sends a standard HTTP POST request to an endpoint exposed by Tyk Gateway (e.g., /chat).
  2. Tyk Streams Role (Producer): Tyk Gateway receives this POST request. An API definition configured with Tyk Streams defines this endpoint as an input. Tyk takes the request payload and publishes it as a message onto a designated backend topic (e.g., the chat topic in Kafka).
  3. A backend service (our “Joker Service”) listens to the chat topic for incoming requests.
• Value Demonstrated:
  • Protocol Bridging: Tyk translates a synchronous HTTP POST into an asynchronous Kafka message.
  • Decoupling: The browser only needs to know the Tyk HTTP endpoint, not the Kafka details (brokers, topic name, protocol).
  • API Management: Tyk can enforce authentication, rate limits, etc., on the /chat endpoint before the message even enters the Kafka system.

​

Tyk as an API Consumer Gateway (Stream to Client)

• Goal: Deliver the asynchronous response (the joke) from the backend system to the client in real time.
• Scenario: The “Joker Service” has processed the request and generated a joke. It needs to send this back to the originating user’s browser session.
• Flow:
  1. The Joker Service publishes the joke response as a message onto a different backend topic (e.g., the discussion topic in Kafka).
  2. Tyk Streams Role (Consumer): Tyk Gateway is configured via another (or the same) API definition to subscribe to the discussion topic.
  3. When Tyk receives a message from the discussion topic, it pushes the message content (the joke) to the appropriate client(s) (provided they have already established a connection) using a suitable real-time protocol like Server-Sent Events (SSE) or WebSockets. Note: In case of multiple clients, events would round-robin amongst the consumers.
• Value Demonstrated:
  • Protocol Bridging: Tyk translates Kafka messages into SSE messages suitable for web clients.
  • Decoupling: The browser doesn’t need a Kafka client; it uses standard web protocols (SSE/WS) provided by Tyk. The Joker Service only needs to publish to Kafka, unaware of the final client protocol.

​

Prerequisites

• Docker: We will run the entire Tyk Stack on Docker. For installation, refer to this guide.
• Git: A CLI tool to work with git repositories. For installation, refer to this guide
• Dashboard License: We will configure Streams API using Dashboard. Contact support to obtain a license.
• Curl and JQ: These tools will be used for testing.

​

Instructions

1. Clone Git Repository: The tyk-demo repository offers a docker-compose environment you can run locally to explore Tyk streams. Open your terminal and clone the git repository using the command below.
   Copy

   Ask AI

   git clone https://github.com/TykTechnologies/tyk-demo
   cd tyk-demo
   

2. Enable Tyk Streams: By default, Tyk Streams is disabled. To enable Tyk Streams in the Gateway and Dashboard, you need to configure the following settings: Create an .env file and populate it with the values below:
   Copy

   Ask AI

   DASHBOARD_LICENCE=<your-dashboard-license>
   GATEWAY_IMAGE_REPO=tyk-gateway-ee
   TYK_DB_STREAMING_ENABLED=true
   TYK_GW_STREAMING_ENABLED=true
   

   • DASHBOARD_LICENCE: Add your license key. Contact support to obtain a license.
   • GATEWAY_IMAGE_REPO: Tyk Streams is available as part of the Enterprise Edition of the Gateway.
   • TYK_DB_STREAMING_ENABLED and TYK_GW_STREAMING_ENABLED: These must be set to true to enable Tyk Streams in the Dashboard and Gateway, respectively. Refer to the configuration options for more details.
3. Start Tyk Streams Execute the following command:
   Copy

   Ask AI

   ./up.sh
   

   This script also starts Kafka within a Docker container, which is necessary for this guide.
   This process will take a few minutes to complete and will display some credentials upon completion. Copy the Dashboard username, password, and API key, and save them for later use.
   Copy

   Ask AI

           ▾ Tyk Demo Organisation
                 Username : [email protected]
                 Password : 3LEsHO1jv1dt9Xgf
         Dashboard API Key : 5ff97f66188e48646557ba7c25d8c601
   

4. Verify Setup: Open Tyk Dashboard in your browser by visiting http://localhost:3000 or http://tyk-dashboard.localhost:3000 and login with the provided admin credentials.
5. Create Producer API: Create a file producer.json with the below content: (Note: tyk-demo-kafka-1 is the hostname used to access Kafka running in a container; alternatively, you can use the IP address assigned to your computer.)

   Show Click to expand the API Definition

   Copy

   Ask AI

       {
           "components": {},
           "info": {
               "title": "jippity-chat",
               "version": "1.0.0"
           },
           "openapi": "3.0.3",
           "paths": {},
           "servers": [
               {
                   "url": "http://tyk-gateway.localhost:8080/jippity-chat/"
               }
           ],
           "x-tyk-api-gateway": {
               "info": {
                   "name": "jippity-chat",
                   "state": {
                       "active": true
                   }
               },
               "server": {
                   "listenPath": {
                       "value": "/jippity-chat/",
                       "strip": true
                   }
               },
               "upstream": {
                 "url": ""
               }
           },
           "x-tyk-streaming": {
               "streams": {
                   "default_stream": {
                       "input": {
                           "http_server": {
                               "address": "",
                               "allowed_verbs": [
                                   "POST"
                               ],
                               "path": "/chat",
                               "rate_limit": "",
                               "timeout": "5s"
                           },
                           "label": ""
                       },
                       "output": {
                           "kafka": {
                               "addresses": ["tyk-demo-kafka-1:9092"],
                               "max_in_flight": 10,
                               "topic": "chat"
                           },
                           "label": ""
                       }
                   }
               }
           }
       }
   

   Create the API by executing the following command. Be sure to replace <your-api-key> with the API key you saved earlier:
   Copy

   Ask AI

   curl -H "Authorization: <your-api-key>" -H "Content-Type: application/vnd.tyk.streams.oas" http://localhost:3000/api/apis/streams -d @producer.json
   

   You should expect a response similar to the one shown below, indicating success. Note that the Meta and ID values will be different each time:
   Copy

   Ask AI

   {"Status":"OK","Message":"API created","Meta":"67e54cadbfa2f900013b501c","ID":"3ddcc8e1b1534d1d4336dc6b64a0d22f"}
   

6. Create Consumer API: Create a file consumer.json with the below content: (Note: tyk-demo-kafka-1 is the hostname used to access Kafka running in a container; alternatively, you can use the IP address assigned to your computer.)

   Show Click to expand the API Definition

   Copy

   Ask AI

       {
           "components": {},
           "info": {
               "title": "jippity-discuss",
               "version": "1.0.0"
           },
           "openapi": "3.0.3",
           "paths": {},
           "servers": [
               {
                   "url": "http://tyk-gateway.localhost:8080/jippity-discuss/"
               }
           ],
           "x-tyk-api-gateway": {
               "info": {
                   "name": "jippity-discuss",
                   "state": {
                       "active": true
                   }
               },
               "server": {
                   "listenPath": {
                       "value": "/jippity-discuss/",
                       "strip": true
                   }
               },
               "upstream": {
                 "url": ""
               }
           },
           "x-tyk-streaming": {
               "streams": {
                   "default_stream": {
                       "input": {
                           "kafka": {
                               "addresses": ["tyk-demo-kafka-1:9092"],
                               "auto_replay_nacks": true,
                               "checkpoint_limit": 1024,
                               "consumer_group": "tyk-streams",
                               "target_version": "3.3.0",
                               "topics": ["discussion"]
                           },
                           "label": ""
                       },
                       "output": {
                           "http_server": {
                               "address": "",
                               "allowed_verbs": [
                                   "GET"
                               ],
                               "stream_path": "/sse"
                           },
                           "label": ""
                       }
                   }
               }
           }
       }
   

   Create the API by executing the following command. Be sure to replace <your-api-key> with the API key you saved earlier:
   Copy

   Ask AI

   curl -H "Authorization: <your-api-key>" -H "Content-Type: application/vnd.tyk.streams.oas" http://localhost:3000/api/apis/streams -d @consumer.json
   

   You should expect a response similar to the one shown below, indicating success. Note that the Meta and ID values will be different each time:
   Copy

   Ask AI

   {"Status":"OK","Message":"API created","Meta":"67e54cadbfa2f900013b501c","ID":"3ddcc8e1b1534d1d4336dc6b64a0d22f"}
   

7. Start Joker Service: Create a file joker-service.sh with the below content:

   Show Click to expand Joker Service

   Copy

   Ask AI

       #!/bin/bash
   
       # Container name
       CONTAINER="tyk-demo-kafka-1"
   
       # Kafka bootstrap server
       BOOTSTRAP_SERVER="localhost:9092"
   
       # Topics
       SOURCE_TOPIC="chat"
       TARGET_TOPIC="discussion"
   
       # Kafka consumer and producer commands
       CONSUMER_CMD="/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server $BOOTSTRAP_SERVER --topic $SOURCE_TOPIC"
       PRODUCER_CMD="/opt/kafka/bin/kafka-console-producer.sh --broker-list $BOOTSTRAP_SERVER --topic $TARGET_TOPIC"
   
       # Joke API URL
       JOKE_API="https://icanhazdadjoke.com/"
   
       echo "Starting to listen for messages on '$SOURCE_TOPIC'..."
   
       # Run the consumer in the container, pipe output to a while loop
       docker exec -i $CONTAINER bash -c "$CONSUMER_CMD" | while IFS= read -r message; do
           # Skip empty lines
           [ -z "$message" ] && continue
   
           echo "Received message from $SOURCE_TOPIC: $message"
   
           # Fetch a random joke from the API and extract it with jq
           joke=$(curl -s -H "Accept: application/json" "$JOKE_API" | jq .joke)
   
           # Check if joke was fetched successfully
           if [ -n "$joke" ]; then
               response_message="In response to '$message': Here's a dad joke - $joke"
           else
               response_message="In response to '$message': Couldn't fetch a joke, sorry!"
           fi
   
           # Send the response message to 'discussion' topic
           echo "$response_message" | docker exec -i $CONTAINER bash -c "$PRODUCER_CMD"
   
           echo "Posted to $TARGET_TOPIC: $response_message"
       done
   
       echo "Consumer stopped."
   

   Make the file executable and start the service.
   Copy

   Ask AI

   chmod +x joker-service.sh
   ./joker-service.sh
   

8. Test the API: Open a terminal and execute the following command to start listening for messages from the Consumer API you created:
   Copy

   Ask AI

   curl -N http://tyk-gateway.localhost:8080/jippity-discuss/sse
   

   In a second terminal, execute the command below to send a message to the Producer API. You can run this command multiple times and modify the message to send different messages:
   Copy

   Ask AI

   curl -X POST http://tyk-gateway.localhost:8080/jippity-chat/chat -H "Content-Type: text/plain" -d "Tell me a joke."
   

   Now, you will see the message appear in the terminal window where you are listening for messages.

​

How It Works

​

High-Level Architecture

• Tyk API Gateway: The core API management platform. It is the entry point, handling initial request processing (like authentication and rate limiting) and routing requests.
• Tyk Streams Module: An integrated extension within the Gateway designed explicitly for asynchronous communication. It intercepts relevant requests and manages the streaming logic.
• Event Brokers / Sources: External systems that act as the origin or destination for data streams. Examples include Apache Kafka, NATS, MQTT brokers, or WebSocket servers. Tyk Streams connects to these systems based on API configuration.
• Upstream Services / APIs: The backend systems, microservices, or APIs that ultimately produce or consume the data being streamed or processed via Tyk Streams.

​

Internal Components of Tyk Streams

1. Stream Middleware: This component plugs into the Tyk Gateway’s request processing chain. It runs after standard middleware like authentication and rate limiting but before the request would normally be proxied. Its job is to inspect incoming requests, identify if they match a configured stream path, and if so, divert them from the standard proxy flow into the stream handling logic.
2. Stream Manager: Acts as the supervisor for streaming operations defined in an API. A given stream configuration is responsible for initializing, managing the lifecycle (starting/stopping), and coordinating the necessary Stream Instances. It ensures the correct streaming infrastructure is ready based on the API definition.
3. Stream Instance: Represents a running, active instance of a specific stream processing task. Each instance executes the logic defined in its configuration – connecting to an event broker, processing messages, transforming data, handling connections, etc. There can be multiple instances depending on the configuration and workload.
4. Stream Analytics: This component captures connection attempts and errors related to HTTP outputs. This data can be exported to popular analytics platforms like Prometheus, OpenTelemetry, and StatsD.

​

Request Processing Flow

1. Request Arrival & Gateway Pre-processing: A client sends a request to an API endpoint managed by Tyk Gateway. The request passes through the initial middleware, such as authentication, key validation, and rate limiting.
2. Streaming Middleware Interception: The request reaches the Stream Middleware. It checks the request path against the stream routes defined in the API configuration.
3. Path Matching:
   • If No Match: The request is not destined for a stream. The Stream Middleware does nothing, and the request continues down the standard Tyk middleware chain, typically being proxied to the configured upstream service.
   • If Match: The request is intended for a stream. The Stream Middleware takes control of the request handling.
4. Stream Manager Coordination: The middleware interacts with the Stream Manager associated with the API’s stream configuration. The Stream Manager ensures the required Stream Instance(s) are initialized and running based on the loaded configuration. This might involve creating a new instance or reusing a cached one.
5. Stream Instance Execution: The instance then executes its defined logic, interacting with the configured Upstream Service / Event Broker (e.g., publishing a message to Kafka, subscribing to an MQTT topic, forwarding data over a WebSocket).
6. Analytics Capture: The Stream Analytics component captures relevant metrics throughout the stream handling process.
7. Final Gateway Response: The response or data stream generated by the streaming components is relayed back through the Gateway to the originating client.

​

Scaling and Availability

​

Configuration Options

1. System-Level Configuration: Enabling the Streams functionality globally within your Tyk Gateway and Tyk Dashboard instances. This activates the necessary components but doesn’t define any specific streams.
2. API-Level Configuration: Defining the actual stream behaviors (inputs, outputs, processing logic) within a specific Tyk OAS API Definition using the x-tyk-streaming extension. This is where you specify how data flows for a particular asynchronous API.

​

System-Level Configuration

​

Tyk Gateway

{
// Partial config from tyk.conf
    "streaming": {
        "enabled": true // Required to activate Streams functionality
    },
// ... more config follows
}

​

Tyk Dashboard

{
// Partial config from tyk_analytics.conf
    "streaming": {
        "enabled": true // Required to activate Streams functionality
    },
// ... more config follows
}

​

API-Level Configuration

• input: Specifies how data enters this stream (e.g., via an HTTP request, by consuming from Kafka, connecting via WebSocket).
• output: Specifies where the data goes after processing (e.g., published to Kafka, sent over WebSocket, delivered via webhook).

{
// Partial config from Tyk OAS API Definition
    "x-tyk-streaming": {
        "streams": {
            "your_stream_name": { // A unique name for this stream configuration within the API
                "input": {
                    // Input configuration object - specifies the data source
                    // Example: "http_server": { ... } or "kafka": { ... }
                },
                "output": {
                    // Output configuration object - specifies the data destination
                    // Example: "kafka": { ... } or "websocket_server": { ... }
                }
                // Optional processing/transformation steps can also be defined here
            },
            "another_stream": { // You can define multiple independent streams
                "input": { ... },
                "output": { ... }
            }
        }
    },
// ... more config follows
}

{
// Partial config from Tyk OAS API Definition
    "x-tyk-streaming": {
        "streams": {
            "http_to_kafka_chat": {
                "input": {
                    "http_server": {
                        "path": "/chat",
                        "allowed_verbs": [ "POST" ],

                    },
                    "label": "HTTP Chat Input"
                },
                "output": {
                    "kafka": {
                        "addresses": ["kafka-broker:9092"],
                        "topic": "chat",

                    },
                    "label": "Kafka Chat Output"
                }
            }
        }
    },
// ... more config follows
}

​

Supported Connectors and Protocols

• Apache Kafka
• WebSocket
• Server-Sent Events (SSE)
• Webhooks

​

Use Cases

​

Security

​

Transformations and Enrichment

• Transformation: Use Tyk’s powerful middleware and plugin system to transform message payloads on the fly. You can convert between different data formats (e.g., JSON to XML), filter fields, or apply custom logic.
• Enrichment: Enrich your async API messages with additional data from external sources. For example, you can lookup customer information from a database and append it to the message payload.

​

Monetization

• Developer Portal Integration: Async APIs can be published to the Tyk Developer Portal, allowing developers to browse, subscribe, and access documentation. Developers can manage their async API subscriptions just like traditional APIs.
• Webhooks: Tyk supports exposing async APIs as webhooks, enabling developers to receive event notifications via HTTP callbacks. Developers can configure their webhook endpoints and subscribe to specific events or topics.

​

Complex Event Processing

• Filter events based on specific criteria
• Aggregate and correlate events from multiple streams
• Enrich events with additional data from other sources
• Detect patterns and sequences of events
• Trigger actions or notifications based on event conditions

input:
  kafka:
    addresses:
      - "localhost:9092" # Replace with actual Kafka broker addresses
    consumer_group: my-group
    topics:
      - orders
output:
  http_server:
    allowed_verbs:
      - GET
    path: /high-value-orders
pipeline:
  processors:
    - mapping: |
        root = if this.order_value > 1000 {
          this
        } else {
          deleted()
        }
    - branch:
        processors:
          - http:
              headers:
                Content-Type: application/json
              url: http://customer-api.local/emails
              verb: POST
        request_map: |-
          root = {
            "customer_id": this.customer_id
          }
        result_map: root.customer_email = this.customer_email
    - mapping: |
        root = this.merge({ "high_value_order": true })

• Tyk Streams Setup: Consumes events from a Kafka topic called orders.
• Processor Block Configuration: Utilizes a custom Mapping script that performs the following operations:
  • Filters orders, only processing those with a value greater than 1000.
  • Enriches the high-value orders by retrieving the customer ID and email from a separate data source.
  • Adds a new high_value_order flag to each qualifying event.
• Output Handling: Processed high-value order events are exposed via a WebSocket stream at the endpoint /high-value-orders.

​

Legacy Modernization

• Decouple legacy systems from modern consumers
• Enable real-time, event-driven communication with legacy apps
• Gradually migrate away from legacy infrastructure

input:
  http_client:
    url: "http://legacy-app/orders"
    verb: GET
    rate_limit: "60s"
pipeline:
  processors:
    - mapping: |
        root.order_id = this.id
        root.total = this.total
        root.timestamp = this.timestamp
output:
  kafka:
    addresses: ["localhost:9092"]
    topic: "orders"

• Tyk Streams periodically polls the legacy /orders REST endpoint every 60 seconds
• The processor transforms the legacy response format into a simplified event structure
• The transformed events are published to a Kafka topic called orders, which can be consumed by modern applications

​

Async API Orchestration

• Combine events from various streams and sources
• Implement complex routing and mediation logic between async APIs
• Create reactive flows triggered by event conditions
• Fanout events to multiple downstream consumers

input:
  broker:
    inputs:
      - kafka:
          addresses: ["localhost:9092"]
          topics: ["stream1"]
          consumer_group: "group1"
      - kafka:
          addresses: ["localhost:9092"]
          topics: ["stream2"]
          consumer_group: "group2"
pipeline:
  processors:
    - switch:
        cases:
          - check: 'meta("kafka_topic") == "stream1"'
            processors:
              - mapping: |
                  root.type = "event_from_stream1"
                  root.details = this
              - branch:
                  processors:
                    - http:
                        url: "http://api1.example.com/process"
                        verb: POST
                        body: '${! json() }'
                  result_map: 'root.api1_response = this'
          - check: 'meta("kafka_topic") == "stream2"'
            processors:
              - mapping: |
                  root.type = "event_from_stream2"
                  root.details = this
              - branch:
                  processors:
                    - http:
                        url: "http://api2.example.com/analyze"
                        verb: POST
                        body: '${! json() }'
                  result_map: 'root.api2_response = this'
    - mapping: 'root = if this.type == "event_from_stream1" && this.api1_response.status == "ok" { this } else if this.type == "event_from_stream2" && this.api2_response.status == "ok" { this } else { deleted() }'
output:
  broker:
    pattern: "fan_out"
    outputs:
      - kafka:
          addresses: ["localhost:9092"]
          topic: "processed_stream1"
          client_id: "tyk_fanout1"
      - kafka:
          addresses: ["localhost:9092"]
          topic: "processed_stream2"
          client_id: "tyk_fanout2"
      - http_client:
          url: "https://webhook.site/unique-id"
          verb: POST
          body: '${! json() }'

1. Input Configuration
   • Uses a broker to combine events from two different Kafka topics, stream1 and stream2, allowing for the integration of events from various streams.
2. Complex Routing and Processing
   • A switch processor directs messages based on their origin (differentiated by Kafka topic metadata).
   • Each stream’s messages are processed and conditionally sent to different APIs.
   • Responses from these APIs are captured and used to decide on message processing further.
3. Reactive Flows
   • Conditions based on API responses determine if messages are forwarded or discarded, creating a flow reactive to the content and success of API interactions.
   • Fanout to Multiple Consumers:
   • The broker output with a fan-out pattern sends processed messages to multiple destinations: two different Kafka topics and an HTTP endpoint, demonstrating the capability to distribute events to various downstream consumers.

​

Monetize APIs using Developer Portal

​

Publishing Async APIs to the Developer Portal

• In the Tyk Dashboard, create a new API and define the async API endpoints and configuration.
• Associate the async API with an API product.
• Publish the API product to the Developer Portal.
• Copy code

​

Async API Documentation

• Overview and purpose of the async API
• Supported protocols and endpoints (e.g., WebSocket, Webhook)
• Event types and payloads
• Subscription and connection details
• Example code snippets for consuming the async API
• Error handling and troubleshooting guidelines

​

Enabling Developers to Subscribe to Events and Streams

1. In the Tyk Developer Portal, navigate to the API product settings.
2. Enable the “Webhooks” option and specify the available events for subscription.
3. Save the API product settings.

• In the Developer Portal, navigate to the My Apps page.
• Select the desired app.
• In the “Webhooks” section, click on “Subscribe”.
• Provide the necessary details:
  • Webhook URL: The URL where the event notifications will be sent.
  • HMAC Secret: Provide a secret key used to sign the webhook messages for authentication.
  • Events: Select the specific events to subscribe to.
• Save the subscription settings.
• Copy code

outputs:
  - portal_webhook:
      event_type: bar
      portal_url: http://localhost:3001
      secret: <portal-api-secret>

​

Glossary

​

Event

​

Stream

​

Publisher (or Producer)

​

Subscriber (or Consumer)

​

Broker

​

Topic (or Channel)

​

FAQ