Add FastStream STOMP broker (#99)

Author: vrslev

.github/workflows/test.yml

@@ -49,19 +49,3 @@ jobs:
           cache-dependency-glob: "**/pyproject.toml"
       - run: uv python install ${{ matrix.python-version }}
       - run: just install test -vv
-
-  test-integration:
-    runs-on: ubuntu-latest
-    strategy:
-      fail-fast: false
-      matrix:
-        python-version:
-          - "3.11"
-          - "3.12"
-          - "3.13"
-    steps:
-      - uses: actions/checkout@v4
-      - uses: extractions/setup-just@v2
-      - run: just test-integration
-        env:
-          PYTHON_IMAGE: python:${{ matrix.python-version }}-slim-bullseye

Dockerfile

@@ -17,5 +17,5 @@ COPY packages/faststream-stomp/faststream_stomp/__init__.py packages/faststream-
 
 ENV SETUPTOOLS_SCM_PRETEND_VERSION=0
 RUN --mount=type=cache,target=~/.cache/uv \
-    ls packages/faststream-stomp && uv lock && uv sync
+    uv lock && uv sync --all-extras --all-packages --frozen
 COPY . .

Justfile

@@ -1,4 +1,4 @@
-default: install lint check-types test test-integration
+default: install lint check-types test
 
 install:
     uv lock --upgrade
@@ -11,13 +11,16 @@ lint:
 check-types:
     uv run mypy .
 
-test *args:
-    uv run pytest {{args}}
+test-fast *args:
+    uv run pytest \
+        --ignore=packages/stompman/test_stompman/test_integration.py \
+        --ignore=packages/faststream-stomp/test_faststream_stomp/test_integration.py {{args}}
 
-test-integration *args:
+test *args:
     #!/bin/bash
     trap 'echo; docker compose down --remove-orphans' EXIT
-    docker compose run --build --rm app .venv/bin/pytest packages/stompman/test_stompman/integration.py --no-cov {{args}}
+    docker compose up -d
+    uv run pytest {{args}}
 
 run-artemis:
     #!/bin/bash

README.md

@@ -130,6 +130,10 @@ Also, I want to pointed out that:
 - stompman is tested and used with [ActiveMQ Artemis](https://activemq.apache.org/components/artemis/) and [ActiveMQ Classic](https://activemq.apache.org/components/classic/).
 - Specification says that headers in CONNECT and CONNECTED frames shouldn't be escaped for backwards compatibility. stompman escapes headers in CONNECT frame (outcoming), but does not unescape headers in CONNECTED (outcoming).
 
+### FastStream STOMP broker
+
+[An implementation of STOMP broker for FastStream.](packages/faststream-stomp/README.md)
+
 ### Examples
 
-See producer and consumer examples in [examples/](examples).
+See examples in [examples/](examples).

conftest.py

@@ -0,0 +1,24 @@
+from typing import cast
+
+import pytest
+import stompman
+
+
+@pytest.fixture(
+    params=[
+        pytest.param(("asyncio", {"use_uvloop": True}), id="asyncio+uvloop"),
+        pytest.param(("asyncio", {"use_uvloop": False}), id="asyncio"),
+    ],
+)
+def anyio_backend(request: pytest.FixtureRequest) -> object:
+    return request.param
+
+
+@pytest.fixture(
+    params=[
+        stompman.ConnectionParameters(host="127.0.0.1", port=9000, login="admin", passcode=":=123"),
+        stompman.ConnectionParameters(host="127.0.0.1", port=9001, login="admin", passcode=":=123"),
+    ]
+)
+def connection_parameters(request: pytest.FixtureRequest) -> stompman.ConnectionParameters:
+    return cast("stompman.ConnectionParameters", request.param)

docker-compose.yml

@@ -1,26 +1,16 @@
 services:
-  app:
-    build:
-      context: .
-      args:
-        PYTHON_IMAGE: ${PYTHON_IMAGE:-python:3.13-slim-bullseye}
-    depends_on:
-      activemq-artemis:
-        condition: service_started
-      activemq-classic:
-        condition: service_started
-
   activemq-artemis:
     image: apache/activemq-artemis:2.37.0-alpine
     environment:
       ARTEMIS_USER: admin
       ARTEMIS_PASSWORD: ":=123"
     ports:
-      - 8161:8161
-      - 61616:61616
+      - 9000:61616
 
   activemq-classic:
     image: apache/activemq-classic:6.1.2
     environment:
       ACTIVEMQ_CONNECTION_USER: admin
       ACTIVEMQ_CONNECTION_PASSWORD: ":=123"
+    ports:
+      - 9001:61613

examples/consumer.py

@@ -2,7 +2,7 @@
 
 import stompman
 
-server = stompman.ConnectionParameters(host="0.0.0.0", port=61616, login="admin", passcode=":=123")  # noqa: S104
+server = stompman.ConnectionParameters(host="127.0.0.1", port=9000, login="admin", passcode=":=123")
 
 
 async def handle_message(message_frame: stompman.MessageFrame) -> None:

examples/faststream_broadcast.py

@@ -0,0 +1,33 @@
+import asyncio
+
+import faststream
+import faststream_stomp
+import stompman
+
+server = stompman.ConnectionParameters(host="127.0.0.1", port=9000, login="admin", passcode=":=123")
+broker = faststream_stomp.StompBroker(stompman.Client([server]))
+
+
+@broker.subscriber("first")
+@broker.publisher("second")
+def _(message: str) -> str:
+    print(message)  # noqa: T201
+    return "Hi from first handler!"
+
+
+@broker.subscriber("second")
+def _(message: str) -> None:
+    print(message)  # noqa: T201
+
+
+app = faststream.FastStream(broker)
+
+
+@app.after_startup
+async def send_first_message() -> None:
+    await broker.connect()
+    await broker.publish("Hi from startup!", "first")
+
+
+if __name__ == "__main__":
+    asyncio.run(app.run())

examples/producer.py

@@ -2,7 +2,7 @@
 
 import stompman
 
-server = stompman.ConnectionParameters(host="0.0.0.0", port=61616, login="admin", passcode=":=123")  # noqa: S104
+server = stompman.ConnectionParameters(host="127.0.0.1", port=9000, login="admin", passcode=":=123")
 
 
 async def main() -> None:

packages/faststream-stomp/README.md

@@ -1,2 +1,50 @@
 # FastStream STOMP broker
 
+## How To Use
+
+Install the package:
+
+```sh
+uv add faststream-stomp
+poetry add faststream-stomp
+```
+
+Basic usage:
+
+```python
+import asyncio
+
+import faststream
+import faststream_stomp
+import stompman
+
+server = stompman.ConnectionParameters(host="127.0.0.1", port=61616, login="admin", passcode="password")
+broker = faststream_stomp.StompBroker(stompman.Client([server]))
+
+
+@broker.subscriber("first")
+@broker.publisher("second")
+def _(message: str) -> str:
+    print(message)  # this will print message from startup
+    return "Hi from first handler!"
+
+
+@broker.subscriber("second")
+def _(message: str) -> None:
+    print(message)  # this will print message from first handler
+
+
+app = faststream.FastStream(broker)
+
+
+@app.after_startup
+async def send_first_message() -> None:
+    await broker.connect()
+    await broker.publish("Hi from startup!", "first")
+
+
+if __name__ == "__main__":
+    asyncio.run(app.run())
+```
+
+Also there are `StompRouter` and `TestStompBroker` for testing. It works similarly to built-in brokers from FastStream, I recommend to read the original [FastStream documentation](https://faststream.airt.ai/latest/getting-started).