No Asyncio NATS

A Python wrapper for NATS that provides an API without requiring asyncio, built with Rust and PyO3.

Rant: This is the real pain of asyncio - libraries need to be implemented twice. WHY? We have gevent, which does not force to reimplement everything.

Internals

This driver uses Rust async_nats under the hood. The Tokio runtime is executed in a seperate (not python aware) thread. Syncronization/Notification between python and Tokio is done via eventfd

Even with this overhead performance is still decent.

By using eventfd this driver can also be used with gevent. Simply use the gevent_connect function, which provides a gevent friendly NATS connection.

TODOS

• publish on pypi
• Implement a .close() method
• Implement error callback on the Client level

Installation

• Go to Actions
• Select the latest green build
• Download the .zip file according to you architecture
• unzip the file
• pip install ./no_asyncio_nats-....whl

Quick Start

import no_asyncio_nats
import datetime

# Connect to NATS
nc = no_asyncio_nats.connect("nats://localhost:4222")

# Subscribe to a subject
sub = nc.subscribe('my.subject')

# Publish a message
nc.publish('my.subject', b'Hello, World!')

# Receive a message
message = sub.recv_msg()
print(f"Received: {message['payload']}")

# Close the connection
sub.unsubscribe()

API Reference

Connection

connect(address, options=None)

Connects to a NATS server and returns a Client instance.

Parameters:

• address (str): The NATS server address (e.g., "nats://localhost:4222")
• options (dict, optional): Connection configuration options

Returns:

• Client: A client instance for interacting with NATS

Connection Options:

The options parameter is a dictionary that can contain the following keys:

Authentication Options:

• token (str): Authentication token
• user_and_password (tuple): Username and password as (username, password)
• nkey (str): Nkey seed for authentication
• credentials_file (str): Path to credentials file
• credentials (str): Credentials as a string

TLS Options:

• tls_root_certificates (str): Path to root certificate file
• tls_client_cert (str): Path to client certificate file
• tls_client_key (str): Path to client private key file
• require_tls (bool): Whether TLS is required
• tls_first (bool): Whether to establish TLS connection before handshake

Timeout and Performance Options:

• ping_interval (datetime.timedelta): Ping interval
• connection_timeout (datetime.timedelta): Connection timeout
• request_timeout (datetime.timedelta): Request timeout
• subscription_capacity (int): Maximum number of subscriptions
• read_buffer_capacity (int): Read buffer size

Connection Behavior Options:

• no_echo (bool): Disable message echo
• custom_inbox_prefix (str): Custom inbox prefix
• name (str): Client name
• retry_on_initial_connect (bool): Retry on initial connection failure
• max_reconnects (int): Maximum number of reconnection attempts
• ignore_discovered_servers (bool): Ignore server discovery
• retain_servers_order (bool): Retain server connection order

Client Class

The main client class for interacting with NATS.

Methods

publish(subject, data, reply=None, headers=None)

Publishes a message to a subject.

Parameters:

• subject (str): The subject to publish to
• data (bytes): The message payload as bytes
• reply (str, optional): Reply subject for responses
• headers (dict, optional): Message headers as key-value pairs

Returns:

• None

flush()

Flushes all pending messages to the server.

Parameters:

• None

Returns:

• None

subscribe(subject)

Subscribes to a subject and returns a Subscriber instance.

Parameters:

• subject (str): The subject to subscribe to

Returns:

• Subscriber: A subscriber instance for receiving messages

queue_subscribe(subject, queue_group)

Subscribes to a subject with queue group semantics.

Parameters:

• subject (str): The subject to subscribe to
• queue_group (str): The queue group name

Returns:

• Subscriber: A subscriber instance for receiving messages

request(subject, data, headers=None)

Performs a request-response pattern.

Parameters:

• subject (str): The subject to send the request to
• data (bytes): The request payload as bytes
• headers (dict, optional): Request headers as key-value pairs

Returns:

• dict: Response message containing:
  • subject (str): Response subject
  • payload (bytes): Response payload
  • reply (str, optional): Reply subject
  • headers (dict, optional): Response headers

new_inbox()

Creates a new unique inbox name.

Parameters:

• None

Returns:

• str: A unique inbox name

Subscriber Class

Represents a subscription to a NATS subject.

Methods

drain()

Drains the subscription, allowing in-flight messages to be processed.

Parameters:

• None

Returns:

• None

unsubscribe()

Unsubscribes from the subject.

Parameters:

• None

Returns:

• None

unsubscribe_after(count)

Unsubscribes after receiving a specific number of messages.

Parameters:

• count (int): Number of messages to receive before unsubscribing

Returns:

• None

recv_msg(timeout=None)

Receives a message from the subscription.

Parameters:

• timeout (datetime.timedelta, optional): Timeout. If None, blocks indefinitely.

Returns:

• dict or None: Message dictionary or None if timeout occurs. Message contains:
  • subject (str): Message subject
  • payload (bytes): Message payload
  • reply (str, optional): Reply subject
  • headers (dict, optional): Message headers

Examples

Basic Publishing and Subscribing

import no_asyncio_nats

# Connect to NATS
nc = no_asyncio_nats.connect("nats://localhost:4222")

# Subscribe to a subject
sub = nc.subscribe('updates')

# Publish a message
nc.publish('updates', b'Hello, World!')

# Receive a message
message = sub.recv_msg()
print(f"Received: {message['payload']}")

# Clean up
sub.unsubscribe()

Request-Response Pattern

import no_asyncio_nats

nc = no_asyncio_nats.connect("nats://localhost:4222")

# Send a request
response = nc.request('service.hello', b'Hello, Service!')
print(f"Response: {response['payload']}")

# Handle multiple requests
def handle_request(message):
    print(f"Received request: {message['payload']}")
    response_data = b"Hello, Client!"
    nc.publish(message['reply'], response_data)

# Set up request handler
sub = nc.subscribe('service.hello')
while True:
    msg = sub.recv_msg()
    handle_request(msg)

Queue Groups

import no_asyncio_nats
import threading

def worker(worker_id):
    nc = no_asyncio_nats.connect("nats://localhost:4222")
    sub = nc.queue_subscribe('tasks', 'worker-group')

    while True:
        msg = sub.recv_msg()
        print(f"Worker {worker_id} processing: {msg['payload']}")

# Start multiple workers
for i in range(3):
    threading.Thread(target=worker, args=(i,)).start()

Advanced Connection Options

import no_asyncio_nats
import datetime

# Connect with authentication and TLS
options = {
    "name": "my-client",
    "user_and_password": ("username", "password"),
    "tls_root_certificates": "/path/to/ca.crt",
    "tls_client_cert": "/path/to/client.crt",
    "tls_client_key": "/path/to/client.key",
    "connection_timeout": datetime.timedelta(seconds=10),  # Use timedelta for time durations
    "ping_interval": datetime.timedelta(seconds=30),      # Use timedelta for time durations
    "max_reconnects": 5
}

nc = no_asyncio_nats.connect("nats://localhost:4222", options)

Error Handling

The API uses exceptions to handle errors. Common exceptions include:

• ConnectionError: When connection to NATS fails
• TimeoutError: When operations timeout
• ValueError: When invalid parameters are provided

Always handle exceptions appropriately:

try:
    nc = no_asyncio_nats.connect("nats://localhost:4222")
    response = nc.request('service.hello', b'Hello')
except ConnectionError as e:
    print(f"Connection failed: {e}")
except TimeoutError as e:
    print(f"Request timed out: {e}")