class Amber::Adapters::MemoryPubSubAdapter

Overview

In-memory implementation of PubSubAdapter.

This adapter provides pub/sub messaging using in-memory channels and fiber-based message routing. This is the default pub/sub adapter and is suitable for development, testing, and single-instance applications.

Note: Messages are only routed within the same application instance. For production applications that require multi-instance deployments or distributed messaging, consider using a Redis-based or message queue adapter.

Usage

# Configure in your application settings
config.pubsub_adapter = Amber::Adapters::MemoryPubSubAdapter.new

Defined in:

amber/adapters/memory_pubsub_adapter.cr

Constructors

Instance Method Summary

Instance methods inherited from class Amber::Adapters::PubSubAdapter

active_topics : Array(String) active_topics, close : Nil close, healthy? : Bool healthy?, publish(topic : String, sender_id : String, message : JSON::Any) : Nil publish, subscribe(topic : String, &block : String, JSON::Any -> Nil) : Nil subscribe, subscriber_count : Int32 subscriber_count, unsubscribe(topic : String) : Nil unsubscribe, unsubscribe_all : Nil unsubscribe_all

Constructor Detail

def self.new #

[View source]

Instance Method Detail

def active_topics : Array(String) #

Lists all active topics that have subscribers.

@return Array of topic names that currently have active subscriptions


[View source]
def clear_all_subscriptions : Nil #

Clears all subscriptions.

This is mainly useful for testing or when shutting down the adapter.


[View source]
def close : Nil #

Closes the adapter and cleans up any resources.


[View source]
def has_subscribers?(topic : String) : Bool #

Checks if there are any subscribers for the given topic.

@param topic The topic to check @return True if there are subscribers, false otherwise


[View source]
def publish(topic : String, sender_id : String, message : JSON::Any) : Nil #

Publishes a message to the specified topic.

The message is immediately delivered to all subscribers of the topic in the current application instance.

@param topic The topic to publish to @param sender_id Unique identifier of the message sender @param message The message payload as JSON::Any


[View source]
def stats : Hash(String, Int32) #

Returns statistics about the current state of the adapter.

@return Hash containing metrics like total topics, total subscribers, etc.


[View source]
def subscribe(topic : String, &block : String, JSON::Any -> Nil) : Nil #

Subscribes to messages on the specified topic.

The listener will be called for each message published to the topic. Multiple listeners can subscribe to the same topic.

@param topic The topic to subscribe to @param block Callback block that receives sender_id and message


[View source]
def subscriber_count(topic : String) : Int32 #

Returns the number of subscribers for a given topic.

@param topic The topic to check @return Number of active subscribers for the topic


[View source]
def unsubscribe(topic : String) : Nil #

Unsubscribes from a topic.

This removes all listeners for the specified topic. If you need to remove specific listeners, you'll need to track them separately.

@param topic The topic to unsubscribe from


[View source]
def unsubscribe_all : Nil #

Unsubscribes from all topics and cleans up any resources.


[View source]