broker = KafkaBroker("localhost:9092")

@broker.publisher(“prediction”)
@broker.subscriber(“input_data”)
async def on_input_data(msg: InputData) -> Prediction:
  # your processing processing
  return prediction

# Just change the broker class, 
#  rest of the code stays the same
broker = RabbitBroker("localhost:5672")

@broker.publisher(“prediction”)
@broker.subscriber(“input_data”)
async def on_input_data(msg: InputData) -> Prediction:
  # your processing processing
  return prediction

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

@broker.subscriber(“input_data”)
async def on_input_data(msg: InputData): # <- decodes consumed message using InputData(**json.loads(data))
  # your processing logic

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

@broker.publisher(“prediction”)
@broker.subscriber(“input_data”)
async def on_input_data(msg: InputData) -> Prediction: # <- encodes produced message using Prediction.json()
  # some processing
  return prediction

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

@broker.subscriber(“input_data”)
# Load a global logger instance from faststream.Context()
async def on_input_data(msg: InputData, logger=Context()):
  logger.info(msg)

async def test_base_app():
  # Subscribe to prediction topic so that we can assert incoming msgs
  @broker.subscriber("prediction")
  async def on_prediction(msg: Prediction):
    pass

  async with TestKafkaBroker(broker):
    # Publish a test message to the input_data topic
    await broker.publish(InputData(data=0.2), "input_data")

    # Check that the handle function for "input_data" topic was called with the correct msg
    on_input_data.mock.assert_called_once_with(dict(InputData(data=0.2)))

    # Check that the service responded with the correct prediction in the "prediction" topic
    on_prediction.mock.assert_called_once_with(dict(Prediction(score=1.2)))

# Create a FastStream router
router = KafkaRouter("localhost:9092")

...

# Connect a FastStream router to a FastAPI application lifespan
app = FastAPI(lifespan=router.lifespan_context)

pip install cookiecutter

cookiecutter https://github.com/airtai/cookiecutter-faststream.git

You`ve downloaded /Users/tvrtko/.cookiecutters/cookiecutter-faststream before. Is it okay to delete and re-download it? [y/n] (y): y
  [1/4] username (github-username): sternakt
  [2/4] project_name (My FastStream App): Greetings App
  [3/4] project_slug (greetings_app): greetings_app
  [4/4] Select streaming_service
    1 - kafka
    2 - nats
    3 - rabbit
    Choose from [1/2/3] (1): 1
```

Change the working directory to the newly created directory and install all development requirements using pip:
```sh
cd greetings_app
pip install -e ".[dev]"
```

Now we are ready to edit our greetings_app/application.py and tests/test_application.py files to implement our application logic. 


### Writing app code

**FastStream** brokers provide convenient function decorators `@broker.subscriber` and `@broker.publisher` to allow you to delegate the actual process of:

- consuming and producing data to Event queues, and

- decoding and encoding JSON encoded messages

These decorators make it easy to specify the processing logic for your consumers and producers, allowing you to focus on the core business logic of your application without worrying about the underlying integration.

Also, **FastStream** uses [**Pydantic**](https://docs.pydantic.dev/) to parse input JSON-encoded data into Python objects, making it easy to work with structured data in your applications, so you can serialize your input messages just using type annotations.

Here is an example python app we talked about:

```python
from faststream import FastStream, Logger
from faststream.kafka import KafkaBroker
from pydantic import BaseModel, Field

version = "0.1.0"
title = "My FastStream service"
description = "Description of my FastStream service"


class Name(BaseModel):
    name: str = Field(..., description="Name of the person")


class Greeting(BaseModel):
    greeting: str = Field(..., description="Greeting message")


broker = KafkaBroker()
app = FastStream(broker, title=title, version=version, description=description)

to_greetings = broker.publisher(
    "greetings",
    description="Produces a message on greetings after receiving a meesage on names",
)


@broker.subscriber("names", description="Consumes messages from names topic and produces messages to greetings topic")
async def on_names(msg: Name, logger: Logger) -> None:
    result = f"hello {msg.name}"
    logger.info(result)
    greeting = Greeting(greeting=result)
    await to_greetings.publish(greeting)
```

The example application will subscribe to **persons** Kafka topic and consume Name JSON messages from it. When the application consumes a message it will publish a Greetings JSON message **greetings** topic.

We can save the application into the `application.py` file and let's take a closer look at the code.

**Creating a broker**
To create an application, first we need to create a broker. This is the main piece of FastStream and takes care of the defining subscribers and producers.

```python
version = "0.1.0"
title = "My FastStream service"
description = "Description of my FastStream service"

...

broker = KafkaBroker()
app = FastStream(broker, title=title, version=version, description=description)
```

**Defining data structures**
Next, we need to define the structure of incoming and outgoing data. FastStream is integrated with Pydantic and offers automatic encoding and decoding of JSON formatted messages into Pydantic classes.

```python
class Name(BaseModel):
    name: str = Field(..., description="Name of the person")


class Greeting(BaseModel):
    greeting: str = Field(..., description="Greeting message")
```

**Defining a publisher**
Now, we define the publishing logic of our application.

```python
to_greetings = broker.publisher(
    "greetings",
    description="Produces a message on greetings after receiving a message on names",
)
```

**Defining a subscriber**
Finally, we can define the subscribing logic of our application. The app will consume data from the "names" topic and use the defined publisher to produce to the "greetings" topic whenever a message is consumed.

```python
@broker.subscriber("names", description="Consumes messages from names topic and produces messages to greetings topic")
async def on_names(msg: Name, logger: Logger) -> None:
    result = f"hello {msg.name}"
    logger.info(result)
    greeting = Greeting(greeting=result)
    await to_greetings.publish(greeting)
```

### Testing the service

The service can be tested using the `TestBroker` context managers, which, by default, puts the Broker into "testing mode".

The Tester will redirect your `subscriber` and `publisher` decorated functions to the InMemory brokers, allowing you to quickly test your app without the need for a running broker and all its dependencies.

Using pytest, the test for our service would look like this:

```python
import pytest
from faststream.kafka import TestKafkaBroker

from greetings_app.application import Greeting, Name, broker, on_names


# Subscribe to the "greetings" topic so we can monitor 
# messages our application is producing
@broker.subscriber("greetings")
async def on_greetings(msg: Greeting) -> None:
    pass


@pytest.mark.asyncio
async def test_on_names():
    async with TestKafkaBroker(broker):
        # Send John to "names" topic
        await broker.publish(Name(name="John"), "names")

        # Assert that our application has consumed "John"
        on_names.mock.assert_called_with(dict(Name(name="John")))

        # Assert that our application has greeted John in the "greetings" topic
        on_greetings.mock.assert_called_with(dict(Greeting(greeting="hello John")))
```

In the test, we send a test User JSON to the **in** topic, and then we assert that the broker has responded to the **out** topic with the appropriate message.

We can save the test to the test_application.py file and run the test by executing the following command in your application root file.

```shell
pytest
```

Here is how the tests execution should look like in your terminal:


```shell
===================================== test session starts =====================================
platform darwin -- Python 3.11.5, pytest-7.4.2, pluggy-1.3.0
rootdir: /Users/tvrtko/Documents/Airt Projects/FastStream/faststream-cookiecutter/greetings_app
configfile: pyproject.toml
plugins: asyncio-0.21.1, anyio-3.7.1
asyncio: mode=Mode.STRICT
collected 1 item                                                                              

tests/test_application.py .                                                             [100%]

====================================== 1 passed in 0.34s ======================================
```

### Running the application

The application can be started using built-in **FastStream** CLI command.

To run the service, use the **FastStream CLI** command and pass the module (in this case, the file where the app implementation is located) and the app symbol to the command.

``` shell
faststream run greetings_app.application:app
```

After running the command, you should see the following output:

``` shell
2023-10-13 08:36:32,162 INFO     - FastStream app starting...
2023-10-13 08:36:32,170 INFO     - names |            - `OnNames` waiting for messages
2023-10-13 08:36:32,177 INFO     - FastStream app started successfully! To exit, press CTRL+C
```

Also, **FastStream** provides you a great hot reload feature to improve your Development Experience

``` shell
faststream run greetings_app.application:app --reload
```

And multiprocessing horizontal scaling feature as well:

``` shell
faststream run greetings_app.application:app --workers 3
```

### Documentation

FastStream provides a command to serve the AsyncAPI documentation, let's use it to document our application.
To generate and serve the documentation, run the following command:

```shell
faststream docs serve greetings_app.application:app
```

Now, you should see the following output:

```shell
INFO:     Started server process [47151]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:8000 (Press CTRL+C to quit)
```

Now open your browser at `http://localhost:8000` and enjoy in your automatically generated documentation! :tada:

![Generated docs](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/aa3d6fjauuqmbdvck6sx.png)

Aaaand, that's it! :tada: :tada: Feel free to experiment further with your application and checkout [the documentation](https://faststream.airt.ai/latest/) for more complex examples.

## Support us on GitHub and join our community :star:

Ready to join the FastStream revolution? Head over to our [GitHub repository](https://github.com/airtai/faststream) and show your support by starring it. By doing so, you'll stay in the loop with the latest developments, updates, and enhancements as we continue to refine and expand FastStream.

Don't forget, we also have an active Discord channel where you can connect with fellow FastStream enthusiasts, ask questions, and share your experiences. Your active participation in our growing community, whether on GitHub or Discord, is invaluable, and we're grateful for your interest and potential contributions. Together, we can make microservices development simpler and more efficient with FastStream.

## Conclusion

FastStream is your go-to tool for efficient microservices development. It simplifies message queues, supports various brokers, and offers Pydantic validation and auto-doc generation.

We're immensely grateful for your interest, and we look forward to your potential contributions. With FastStream in your toolkit, you're prepared to conquer the challenges of data-centric microservices like never before. Happy coding!