Skip to main content

Fabric Message Bus Schema

Project description

PyPI

Message Bus Schema and Message Definition

Basic framework for a Fabric Message Bus Schema and Messages for Inter Actor Communication

Overview

Fabric communication across various actors in Control and Measurement Framework is implemented using Apache Kafka.

Apache Kafka is a distributed system designed for streams. It is built to be fault-tolerant, high-throughput, horizontally scalable, and allows geographically distributed data streams and stream processing applications.

Kafka enables event driven implementation of various actors/services. Events are both a Fact and a Trigger. Each fabric actor will be a producer for one topic following the Single Writer Principle and would subscribe to topics from other actors for communication. Messages are exchanged over Kafka using Apache Avro data serialization system.

Requirements

  • Python 3.7+
  • confluent-kafka
  • confluent-kafka[avro]

Installation

$ pip3 install .

Usage

This package implements the interface for producer/consumer APIs to push/read messages to/from Kafka via Avro serialization.

Message and Schema

User is expected to inherit IMessage class(message.py) to define it's own members and over ride to_dict() function. It is also required to define the corresponding AVRO schema pertaining to the derived class. This new schema shall be used in producer and consumers.

Example schema for basic IMessage class is available in (schema/message.avsc)

Producers

AvroProducerApi class implements the base functionality for an Avro Kafka producer. User is expected to inherit this class and override delivery_report method to handle message delivery for asynchronous produce.

Example for usage available at the end of producer.py

Consumers

AvroConsumerApi class implements the base functionality for an Avro Kafka consumer. User is expected to inherit this class and override process_message method to handle message processing for incoming message.

Example for usage available at the end of consumer.py

Admin API

AdminApi class provides support to carry out basic admin functions like create/delete topics/partions etc.

How to bring up a test Kafka cluster to test

Generate Credentials

You must generate CA certificates (or use yours if you already have one) and then generate a keystore and truststore for brokers and clients.

cd $(pwd)/secrets
./create-certs.sh
(Type yes for all "Trust this certificate? [no]:" prompts.)
cd -

Set the environment variable for the secrets directory. This is used in later commands. Make sure that you are in the MessageBus directory.

export KAFKA_SSL_SECRETS_DIR=$(pwd)/secrets

Bring up the containers

You can use the docker-compose.yaml file to bring up a simple Kafka cluster containing

  • broker
  • zookeeper
  • schema registry

Use the below command to bring up the cluster

docker-compose up -d

This should bring up following containers:

docker ps
CONTAINER ID        IMAGE                                    COMMAND                  CREATED             STATUS              PORTS                                                                                        NAMES
189ba0e70b97        confluentinc/cp-schema-registry:latest   "/etc/confluent/dock…"   58 seconds ago      Up 58 seconds       0.0.0.0:8081->8081/tcp                                                                       schemaregistry
49616f1c9b0a        confluentinc/cp-kafka:latest             "/etc/confluent/dock…"   59 seconds ago      Up 58 seconds       0.0.0.0:9092->9092/tcp, 0.0.0.0:19092->19092/tcp                                             broker1
c9d19c82558d        confluentinc/cp-zookeeper:latest         "/etc/confluent/dock…"   59 seconds ago      Up 59 seconds       2888/tcp, 0.0.0.0:2181->2181/tcp, 3888/tcp                                                   zookeeper

Project details


Release history Release notifications | RSS feed

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

fabric-message-bus-1.5.0b6.tar.gz (44.8 kB view details)

Uploaded Source

Built Distribution

fabric_message_bus-1.5.0b6-py3-none-any.whl (142.4 kB view details)

Uploaded Python 3

File details

Details for the file fabric-message-bus-1.5.0b6.tar.gz.

File metadata

  • Download URL: fabric-message-bus-1.5.0b6.tar.gz
  • Upload date:
  • Size: 44.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: python-requests/2.28.2

File hashes

Hashes for fabric-message-bus-1.5.0b6.tar.gz
Algorithm Hash digest
SHA256 d243f828f73e9f63ede1ee8f1cebfadef7234e81e6cf6246768ce271f9620ccf
MD5 e63c10009c931b9353091119faf6de54
BLAKE2b-256 91e40be10f89fe9984b5cc4ad80d6f13625f2f3ad770b9ad65e72af66c27e45c

See more details on using hashes here.

File details

Details for the file fabric_message_bus-1.5.0b6-py3-none-any.whl.

File metadata

File hashes

Hashes for fabric_message_bus-1.5.0b6-py3-none-any.whl
Algorithm Hash digest
SHA256 79ba63df51123881a048ee82ca2a84ddd6840a5981898464c9aa3e88ee9943da
MD5 9ad3da1b05cc7c358419c2edda6d9d0d
BLAKE2b-256 209a225a59cddc9573d7b8dcb8a4418c9ad2a954d495f340a61181cd30e903ad

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page