# Event Brokers

Rasa Core allows you to stream events to a message broker. The event broker emits events into the event queue. It becomes part of the `TrackerStore` which you use when starting an `Agent` or launching `rasa.core.run`.

All events are streamed to the broker as serialized dictionaries every time the tracker updates its state. An example event emitted from the `default` tracker looks like this:

```
{
    "sender_id": "default",
    "timestamp": 1528402837.617099,
    "event": "bot",
    "text": "what your bot said",
    "data": "some data"
}
```

The `event` field takes the event’s `type_name` (for more on event types, check out the [Events](https://legacy-docs-v1.rasa.com/1.2.9/api/events/#events) docs).

Rasa enables three possible broker types:

- [Pika Event Broker](https://legacy-docs-v1.rasa.com/1.2.9/api/event-brokers/#pika-event-broker)
- [Kafka Event Broker](https://legacy-docs-v1.rasa.com/1.2.9/api/event-brokers/#kafka-event-broker)
- [SQL Event Broker](https://legacy-docs-v1.rasa.com/1.2.9/api/event-brokers/#sql-event-broker)

## Pika Event Broker

The example implementation uses [Pika](https://pika.readthedocs.io/), the Python client library for [RabbitMQ](https://www.rabbitmq.com/).

### Adding a Pika Event Broker Using the Endpoint Configuration

You can use an endpoint configuration file to instruct Rasa Core to stream all events to your event broker. To do so, add the following section to your endpoint configuration, e.g. `endpoints.yml`:

```
event_broker:
  url: localhost
  username: username
  password: password
  queue: queue
  type: pika
```

Then instruct Rasa Core to use the endpoint configuration by adding `--endpoints <path to your endpoint configuration` as follows:

```
rasa run -m models --endpoints endpoints.yml
```

### Adding a Pika Event Broker in Python

Here is how you add it using Python code:

```
from rasa.core.event_brokers.pika_producer import PikaProducer
from rasa_platform.core.tracker_store import InMemoryTrackerStore

pika_broker = PikaProducer('localhost',
                            'username',
                            'password',
                            queue='rasa_core_events')

tracker_store = InMemoryTrackerStore(db=db, event_broker=pika_broker)
```

### Implementing a Pika Event Consumer

You need to have a RabbitMQ server running, as well as another application that consumes the events. Here’s a simple example:

```
import json
import pika

def _callback(self, ch, method, properties, body):
        print('Received event {}'.format(json.loads(body)))

if __name__ == '__main__':
    credentials = pika.PlainCredentials('username', 'password')
    connection = pika.BlockingConnection(
        pika.ConnectionParameters('rabbit', credentials=credentials))

channel = connection.channel()
    channel.basic_consume(_callback,
                          queue='rasa_core_events',
                          no_ack=True)
    channel.start_consuming()
```

## Kafka Event Broker

It is possible to use [Kafka](https://kafka.apache.org/) as the main broker for your events. The following implementation shows an example of how to instantiate a Kafka producer:

```
from rasa.core.event_brokers.kafka_producer import KafkaProducer
from rasa.core.tracker_store import InMemoryTrackerStore

kafka_broker = KafkaProducer(host='localhost:9092',
                             topic='rasa_core_events')

tracker_store = InMemoryTrackerStore(event_broker=kafka_broker)
```

### Authentication and authorization

Rasa Core’s Kafka producer accepts two types of security protocols - `SASL_PLAINTEXT` and `SSL`.

For development, use simple authentication with `SASL_PLAINTEXT`:

```
kafka_broker = KafkaProducer(host='kafka_broker:9092',
                             sasl_plain_username='kafka_username',
                             sasl_plain_password='kafka_password',
                             security_protocol='SASL_PLAINTEXT',
                             topic='rasa_core_events')
```

For production, use SSL protocol to secure data:

```
kafka_broker = KafkaProducer(host='kafka_broker:9092',
                             ssl_cafile='CARoot.pem',
                             ssl_certfile='certificate.pem',
                             ssl_keyfile='key.pem',
                             ssl_check_hostname=True,
                             security_protocol='SSL',
                             topic='rasa_core_events')
```

### Implementing a Kafka Event Consumer

The following code shows an example for consuming events from Kafka:

```
from kafka import KafkaConsumer
from json import loads

consumer = KafkaConsumer('rasa_core_events',
                          bootstrap_servers=['localhost:29093'],
                          value_deserializer=lambda m: json.loads(m.decode('utf-8')))

for message in consumer:
    print(message.value)
```

## SQL Event Broker

You can use an SQL database as an event broker. Connections are established using [SQLAlchemy](https://www.sqlalchemy.org/).

### Adding a SQL Event Broker Using the Endpoint Configuration

To save all events to your SQL event broker, add a `event_broker` section to your endpoint configuration:

SQLite example:
```
event_broker:
  type: SQL
  dialect: sqlite
  db: events.db
```

PostgreSQL example:
```
event_broker:
  type: SQL
  host: 127.0.0.1
  port: 5432
  dialect: postgresql
  username: myuser
  password: mypassword
  db: mydatabase
```

With this configuration, Rasa will create a table called `events` where all events will be added.