npm package discovery and stats viewer.

Discover Tips

  • General search

    [free text search, go nuts!]

  • Package details

    pkg:[package-name]

  • User packages

    @[username]

Sponsor

Optimize Toolset

I’ve always been into building performant and accessible sites, but lately I’ve been taking it extremely seriously. So much so that I’ve been building a tool to help me optimize and monitor the sites that I build to make sure that I’m making an attempt to offer the best experience to those who visit them. If you’re into performant, accessible and SEO friendly sites, you might like it too! You can check it out at Optimize Toolset.

About

Hi, 👋, I’m Ryan Hefner  and I built this site for me, and you! The goal of this site was to provide an easy way for me to check the stats on my npm packages, both for prioritizing issues and updates, and to give me a little kick in the pants to keep up on stuff.

As I was building it, I realized that I was actually using the tool to build the tool, and figured I might as well put this out there and hopefully others will find it to be a fast and useful way to search and browse npm packages as I have.

If you’re interested in other things I’m working on, follow me on Twitter or check out the open source projects I’ve been publishing on GitHub.

I am also working on a Twitter bot for this site to tweet the most popular, newest, random packages from npm. Please follow that account now and it will start sending out packages soon–ish.

Open Software & Tools

This site wouldn’t be possible without the immense generosity and tireless efforts from the people who make contributions to the world and share their work via open source initiatives. Thank you 🙏

© 2024 – Pkg Stats / Ryan Hefner

mqrpc

v1.5.0

Published

💫 Easy RPC over RabbitMQ

Downloads

15

Readme

MQRPC

💫 Easy RPC over RabbitMQ

Easily implement RPC over your RabbitMQ broker with a few lines of code:

import { RpcClient, RpcServer } from 'mqrpc'

const server = new RpcServer({ amqpClient: { amqpUrl: 'amqp://localhost '} })
const client = new RpcClient({ amqpClient: { amqpUrl: 'amqp://localhost '} })

server.register('math.add', (a, b) => a + b)

await server.init()
await client.init()

await client.call('math.add', 2, 2) // 4

MQRPC leverages RabbitMQs Direct reply-to functionality to implement an RPC system with reasonable defaults that work out-of-the box. All you need is a RabbitMQ broker and its URL.

Features

  • Any number of servers & clients
  • Argument & Return serialization
  • Error serialization
  • Timeout management

API

Both Server & Client are designed to be simple to use and thus have a low-surface API. The following type definitions follow TypeScript syntax loosely:

RpcServer

new RpcServer({ amqpClient: AmqpOpts, rpcServer?: ServerOpts })

Instances a new server with the given config. amqpClient is required:

type AmqpOpts = {
  connection?: amqplib.Connection // Pass a live amqplib connection here to re-use it.
  amqpUrl?: string                // The RabbitMQ URL. Ignored if `connection` is provided.
  socketOptions?: object          // Customize connection to RabbitMQ.
  prefetchCount: number           // Customize consumer prefetch count.
}

type ServerOpts = {
  rpcExchangeName?: string        // Exchange name for server, defaults to 'mqrpc'.
  logger?: object                 // For custom logger injection.
}

Although all configs are optional, one of amqpClient.connection or amqpClient.amqpUrl must be passed.

async server.init()

Declares all the exchanges, queues and bindings for the server. Starts listening for calls from clients, so you should call this after registering all available procedures.

server.register(procedure: string, handler: Function)

Registers a new procedure and its handler in the server. The handler can be synchronous or return a Promise.

server.register('util.echo', arg => arg)
server.register('time.now', () => Date.now())
server.register('math.average', (...args) => args.reduce((acc, val) => acc + val) / args.length)
server.register('meaning.of.life', async () => 42)

register should be called before init to ensure the server won't receive any unknown calls by clients that are already live.

server.registerDebugProcedures()

Adds a default procedure for debugging, with name mqrpc.echo, that returns any given argument.

async server.term()

Neatly shut down the server. Closes the AMQP channel and, if one wasn't provided, the connection as well.

RpcClient

new RpcClient({ amqpClient: AmqpOpts, rpcClient?: ClientOpts })

Instances a new client with the given config. amqpClient is required:

type AmqpOpts = {
  connection?: amqplib.Connection // Pass a live amqplib connection here to re-use it.
  amqpUrl?: string                // The RabbitMQ URL. Ignored if `connection` is provided.
  socketOptions?: object          // Customize connection to RabbitMQ.
  prefetchCount: number           // Customize consumer prefetch count.
}

type ClientOpts = {
  rpcExchangeName?: string        // Exchange name for server, defaults to 'mqrpc'.
  logger?: object                 // For custom logger injection.
  ackTimeout?: number             // How long should the client wait for a Server to start working on a call. Default 0 (no timeout).
  idleTimeout?: number            // How long can the server be idle until it is considered "dead". Default 0 (no timeout).
  callTimeout?: number            // Maximum time from making a call to receiving a reply. Default 900 000 (15 minutes).
}

Although all configs are optional, one of amqpClient.connection or amqpClient.amqpUrl must be passed. Every timeout is in milliseconds and will throw TimeoutExpired when breached. See timeouts below for more info.

async client.init()

Connects to RabbitMQ and starts listening for replies from servers.

async client.call(procedure: string, ...args: any[])

Calls the named procedure on an RpcServer with the given args. Resolves to whatever the procedure returns. Rejects if the procedure throws, or there is a connection error or an error in the server.

Errors

The following error types may be thrown:

  • ProcedureFailed: The most common (hopefully), is thrown when the procedure itself throws. The remote error stack may be included in error.causeStack.
  • ServerError: When an error occurs in the server while processing the call. Eg: the requested procedure is not registered.
  • UnparseableContent: Whatever reply we got from a server could not be parsed.
  • UnknownReply: Reply was parseable, but the format isn't understood.
async client.term({ waitForCalls: number })

Neatly shut down the client. Terminates any active calls, closes the AMQP channel and, if one wasn't provided, the connection as well.

Will wait up to waitForCalls milliseconds for pending calls to resolve, or indefinitely if given 0ms.

Timeouts

Since it may not be sensible to wait forever for a call to resolve, the client exposes three configurable timeouts that will interrupt a call when expired. These are:

ackTimeout

When a server receives a procedure call it will send an ack message back to the client, immediately before beginning execution. This signals the client that a server is handling their call. This timeout signals how long to wait until the ack is received.

This timeout is disabled by default, since it's sensible to expect a server will eventually pick up a client's call. However, it may be used to control for times of high message congestion, for example.

idleTimeout

While the server is executing a procedure, it'll periodically send wait messages back to the client (behind the scenes). This works as a heartbeat of sorts and indicates to the client that the server hasn't crashed, or in some way disconnected. This timeout indicates how long a server may be silent before aborting the call.

This timeout is disabled by default, since RabbitMQ has its own hearbeat functionality that, in conjunction with its own ack mode, guarantees at-least-once execution. You may enable this if operating in noAck mode.

callTimeout

The overall maximum time a call may take to resolve a request. This timeout starts on a procedure call and terminates when a reply is received.

This timeout is 15 minutes by default.

Future Features

  • Publisher drain management
  • Server-side timeout management

Testing

You'll need a local RabbitMQ broker to run the tests. Optionally set env var RABBITMQ_VHOST to specify a vhost, uses / by default. Then:

$ yarn test

Contributing

Feel free to submit PRs. Please include unit tests for any new features.

Why TypeScript

Because I wanted to try it out ¯\(ツ)