ActiveMQClient (activemq_client v0.0.3)

View Source

Generic ActiveMQ client using STOMP protocol for Artemis ActiveMQ.

This library provides a high-level interface for connecting to and interacting with Apache ActiveMQ Artemis using the STOMP protocol. It handles message publishing, consuming, and subscription management with configurable message filtering.

Features

  • STOMP-based connection to Artemis ActiveMQ
  • Message publishing and consuming
  • Address subscription management with ANYCAST (queue) and MULTICAST (topic) routing
  • Configurable message filtering with selectors
  • Automatic reconnection and error handling
  • Callback-based message processing

Usage

# Start the client
{:ok, _} = ActiveMQClient.start_link([
  host: "localhost", 
  port: 61613,
  username: "artemis",
  password: "artemis",
  message_handler: &MyApp.handle_message/1
])

# Publish a message
ActiveMQClient.publish_message("Hello World!", "/queue/test")

# Subscribe to a queue (ANYCAST routing)
ActiveMQClient.subscribe_to_queue("/queue/test")

# Subscribe to address with MULTICAST routing and message filter
ActiveMQClient.subscribe_to_address_with_selector(
  "/topic/events", 
  "type = 'MY_EVENT'", 
  "my_subscription"
)

Summary

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

get_status()

Gets the current client status.

publish_message(message, destination \\ nil)

Publishes a message to an address.

reconnect()

Triggers a manual reconnection attempt.

send_reply(reply_json, reply_destination, message_type)

Sends a reply message to a destination with message type.

set_message_handler(handler)

Sets a custom message handler function.

start_link(opts \\ [])

Starts the ActiveMQ client process.

subscribe_multiple(subscriptions)

Subscribes to multiple addresses with different selectors.

subscribe_to_address_with_selector(address, selector, subscription_id)

Subscribes to an address with a message selector.

subscribe_to_multicast_address(address \\ nil)

Subscribes to an address with MULTICAST routing (topic semantics).

subscribe_to_queue(queue \\ nil)

Subscribes to an address with ANYCAST routing (queue semantics).

unsubscribe(destination)

Unsubscribes from an address.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

get_status()

Gets the current client status.

Returns

Map containing connection status, host, port, subscriptions, etc.

publish_message(message, destination \\ nil)

Publishes a message to an address.

Parameters

  • message - Message content as string
  • destination - Destination address with ANYCAST or MULTICAST routing (optional, uses default if nil)

Returns

  • :ok on success
  • {:error, reason} on failure

reconnect()

Triggers a manual reconnection attempt.

send_reply(reply_json, reply_destination, message_type)

Sends a reply message to a destination with message type.

Parameters

  • reply_json - JSON reply content
  • reply_destination - Destination to send reply to
  • message_type - Message type header

Returns

  • :ok on success

set_message_handler(handler)

Sets a custom message handler function.

Parameters

  • handler - Function to handle incoming messages

start_link(opts \\ [])

Starts the ActiveMQ client process.

Options

  • :host - ActiveMQ server hostname (default: "localhost")
  • :port - ActiveMQ STOMP port (default: 61613)
  • :username - Authentication username (default: "artemis")
  • :password - Authentication password (default: "artemis")
  • :queue - Default queue destination (default: "/queue/artemis_queue")
  • :multicast_address - Default multicast address destination (default: "/topic/artemis_topic")
  • :message_handler - Custom message handler function

subscribe_multiple(subscriptions)

Subscribes to multiple addresses with different selectors.

Parameters

  • subscriptions - List of {address, selector, subscription_id} tuples

Returns

  • :ok on success
  • {:error, reason} on failure

subscribe_to_address_with_selector(address, selector, subscription_id)

Subscribes to an address with a message selector.

Parameters

  • address - Address destination (can be ANYCAST or MULTICAST)
  • selector - Message selector for filtering
  • subscription_id - Unique subscription identifier

Returns

  • :ok on success
  • {:error, reason} on failure

subscribe_to_multicast_address(address \\ nil)

Subscribes to an address with MULTICAST routing (topic semantics).

Parameters

  • address - Address with MULTICAST routing (optional, uses default if nil)

Returns

  • :ok on success
  • {:error, reason} on failure

subscribe_to_queue(queue \\ nil)

Subscribes to an address with ANYCAST routing (queue semantics).

Parameters

  • queue - Address with ANYCAST routing (optional, uses default if nil)

Returns

  • :ok on success
  • {:error, reason} on failure

unsubscribe(destination)

Unsubscribes from an address.

Parameters

  • address - Address to unsubscribe from

Returns

  • :ok on success
  • {:error, reason} on failure