rascal

Prefetch

prefetch limits the number of unacknowledged messages a subscription can have outstanding. This prevents overloading your event loop or downstream services. The default is 10.

Channel Prefetch

channelPrefetch operates at the channel level. Because Rascal uses a dedicated channel per subscriber, this is rarely needed and has higher overhead in clusters. It is primarily used to adjust prefetch dynamically without cancelling the subscription.

Note: When using channelPrefetch, you should set the regular consumer prefetch to 0 to prevent conflicts.

// Dynamically tuning channel prefetch
const subscription = await broker.subscribe('s1', { prefetch: 0, channelPrefetch: 5 });
subscription.on('message', (message, content, ackOrNack) => {
  ackOrNack();
  const prefetch = tunePrefetch();
  await subscription.setChannelPrefetch(prefetch);
});

Rascal extends standard RabbitMQ concepts (Brokers, Vhosts, Exchanges, Queues, Channels, and Connections) with two primary abstractions: Publications and Subscriptions.

• Publication: A named configuration for publishing messages. It defines the destination (queue or exchange), routing configuration, encryption profiles, reliability guarantees, and message options.
• Subscription: A named configuration for consuming messages. It defines the source queue, encryption profiles, content encoding, and delivery options (such as acknowledgement handling and prefetch).

Both must be defined in your configuration and supplied when creating the Rascal broker. Once the broker is initialized, you retrieve these named configurations to interact with the broker.

To allow a consumer to send a reply to the specific application instance that published a message, use the replyTo property.

When replyTo is set to true in a queue configuration, Rascal appends a unique UUID to the queue name. When combined with the replyTo property in a publication configuration, Rascal automatically sets the replyTo property on outbound messages to that unique queue name.

{
  "queues": {
    "q1": {
      "replyTo": true
    }
  },
  "publications": {
    "exchange": "e1",
    "replyTo": "q1"
  }
}

Rascal supports a shorthand notation for configuration to reduce verbosity. You can use arrays of strings for simple definitions or a mix of strings and objects when specific parameters (like exchange types) are required.

// Shorthand for exchanges, queues, and bindings
{
  "exchanges": ["e1", "e2"],
  "queues": ["q1", "q2"],
  "bindings": ["e1 -> q1", "e2[bk1, bk2] -> q2"]
}

// Mixing shorthand with object configuration for specific parameters
{
  "exchanges": [
    "e1",
    {
      "name": "e2",
      "type": "fanout"
    }
  ]
}

Since AMQP doesn't support checking if a vhost exists, Rascal uses the RabbitMQ Management API. This is primarily useful in test environments.

You can specify management details within the connection object. You can also provide a custom agent (e.g., for TLS) via the broker components during creation.

const https = require('https');
const agent = new https.Agent(options);
const components = { agent };
const broker = await Broker.create(config, components);

{
  "vhosts": {
    "v1": {
      "connection": {
        "hostname": "broker.example.com",
        "user": "bob",
        "password": "secret",
        "management": {
          "protocol": "https",
          "pathname": "prefix",
          "user": "admin",
          "password": "super-secret",
          "options": {
            "timeout": 1000
          }
        }
      }
    }
  }
}

To use Rascal with the traditional callback pattern, import Broker from the rascal package. The create method accepts a configuration object and a callback function that receives the broker instance or an error.

const Broker = require('rascal').Broker;
const config = require('./config');

Broker.create(config, (err, broker) => {
  if (err) throw err;

  broker.on('error', console.error);

  // Publish a message
  broker.publish('demo_publication', 'Hello World!', (err, publication) => {
    if (err) throw err;
    publication.on('error', console.error);
  });

  // Consume a message
  broker.subscribe('demo_subscription', (err, subscription) => {
    if (err) throw err;
    subscription
      .on('message', (message, content, ackOrNack) => {
        console.log(content);
        ackOrNack();
      })
      .on('error', console.error);
  });
});

Rascal uses a channel pool that pauses when the connection to the broker is lost. Instead of erroring, publishes are held in an in-memory queue. To prevent messages from being sent once the connection is re-established, listen for the paused event and call publication.abort().

broker.publish('p1', 'some message', (err, publication) => {
  if (err) throw err;
  publication
    .on('success', (messageId) => {
      console.log('Message id was: ', messageId);
    })
    .on('error', (err, messageId) => {
      console.error('Error was: ', err.message);
    })
    .on('paused', (messageId) => {
      console.warn('Publication was paused. Aborting message: ', messageId);
      publication.abort();
    });
});

To run the project tests, you need a RabbitMQ server running locally. You can use the provided Docker command to simplify setup.

npm test
# Or use Docker for a local RabbitMQ instance:
npm run docker

Rascal provides helper methods to merge your custom definitions with sensible defaults. Use withDefaultConfig for production-ready settings (optimized for reliability) or withTestConfig for test environments.

var rascal = require('rascal');
var definitions = require('./your-config.json');

// For production/standard use
var config = rascal.withDefaultConfig(definitions);

// For test environments
var config = rascal.withTestConfig(definitions);