AMQP message queues Preview

Two client objects, one for each incompatible generation of the protocol.

Constructor	Protocol	Brokers
new AmqpClient091()	AMQP 0.9.1	RabbitMQ, Apache Qpid, StormMQ
new AmqpClient10()	AMQP 1.0	Apache ActiveMQ, Azure Service Bus, Azure Event Hubs, Solace

Picking the wrong one just fails to connect

AMQP 0.9.1 and AMQP 1.0 share a name and nothing else. If Connect() returns false against a broker you are certain is up and reachable, the version is the first thing to check.

Everything except the constructor name is identical between the two. Both speak amqp:// and amqps://.

Properties

var cli = new AmqpClient091();
cli.URL  = "amqps://broker.example.com:5671";
cli.User = "connector";
cli.Pass = brokerPassword;

That is the whole surface. There is no PassFromSecret here; some Syncplify Server documentation mentions one, and it does not exist in this engine. Read the password out of the secret store instead:

cli.Pass = GetSecret("broker-password");

Posting a message

This is what a Connector script should be doing with AMQP: firing an event onto a queue and moving on.

var cli = new AmqpClient091();
cli.URL  = "amqps://broker.example.com:5671";
cli.User = "connector";
cli.Pass = brokerPassword;

if (!cli.Connect()) {
    Log.Error("could not reach the AMQP broker");
    Exit(0);           // an after handler: log it and let the operation stand
}

var ok = cli.PostMessage("file.events", JSON.stringify({
    event: EventHandler(),
    path: CtxRelPath(),
    user: CtxUsername(),
    vfs: CtxVFSName(),
    at: FormatDateTime("YYYY-MM-DD HH:mm:ss")
}));

cli.Close();
Log.Info("posted to the queue: " + ok);

PostMessage(queueName: string, message: string): boolean;

Client	How it publishes
AmqpClient091	Through the default exchange, with the queue name as the routing key. The queue is declared if it does not exist
AmqpClient10	Opens a sender link to the address, sends, closes the sender

Reading messages

MonitorQueue(queueName: string): void;
GetMessages(): Msg[];

MonitorQueue() starts a subscription; GetMessages() returns whatever has arrived since you last asked, as an array:

{
  ReceivedAt: Date,
  Queue: string,
  Message: string
}

The field names are capitalized

ReceivedAt, Queue and Message, not receivedAt, queue and message. Some Syncplify Server documentation shows the lowercase spellings; reading those here gives you undefined with no error.

A polling loop does not belong in a Connector script

Consuming a queue means looping until something arrives, and a Connector script does not get to do that. Every execution is bounded by the handler timeout, five seconds by default, and the loop takes the whole handler down with it when the deadline passes. Worse, HaltSignalReceived() and ConsoleFeedback, which Syncplify AFT scripts use to break out of exactly this kind of loop, do not exist here. A Syncplify Server or AFT queue monitoring example pasted into a Connector script will not run.

A Connector script is a reaction to a file event. It should post to a queue and finish. If you need something that consumes a queue continuously, that is a service of yours running beside the Connector, not a script inside it.

If you nonetheless need to collect a reply within one execution, bound it explicitly and keep it well inside the handler's budget:

cli.MonitorQueue("responses");
cli.PostMessage("requests", JSON.stringify({ action: "check", path: CtxRelPath() }));

var replies = [];
var deadline = 3;                       // seconds, against a handler timeout well above that
for (var i = 0; i < deadline * 2; i++) {
    Sleep(500);
    replies = cli.GetMessages();
    if (replies.length > 0) { break; }
}
cli.Close();

if (replies.length === 0) {
    Log.Warn("no reply from the queue in time");
    Exit(0);
}
Log.Info("reply: " + replies[0].Message);

Even this belongs on an after handler with a generous timeout, never in front of a transfer.

Closing

Connect(): boolean;
Close(): boolean;

Close every client you open. A Connector that opens a broker connection per upload and never closes one exhausts the broker's connection limit quickly.