Getting AMQP to work in your browser
For one of my customers, I had to be able to connect to an EventHub through the browser, so how did I do this? So we know that EventHub works with the AMQP protocol, so what if we could get this working in the frontend?
After fiddling a bit with different kind of solutions, I went with a quick and dirty way to get this working. To start, let's download the RHEA library which allows us to send messages using the AMQP protocol https://github.com/amqp/rhea. This library however is designed to work with Node.js, so we will have to make some adaptations to get this working in our browser using Browserify.
Changes to get RHEA working in a browser context
lib/types.js
} else if (o instanceof Buffer) {
return types.wrap_binary(o);
}
change to
} else if (o instanceof Buffer || o instanceof ArrayBuffer) {
return types.wrap_binary(o);
}
lib/container.js
Add the following code somewhere.
Container.prototype.disconnect = function () {
this.connection.close();
};
Modify:
Container.prototype.connect = function (options) {
return new Connection(options, this).connect();
};
to
Container.prototype.connect = function (options) {
this.connection = new Connection(options, this);
return this.connection.connect();
};
Build
# Note: this might fail on the tests, ignore it
npm install
# Note: this will output rhea.js in the ./dist folder
npm run browserify
Connecting to an EventHub
To connect to our EventHub, feel free to use the code below:
// [eventHubPath]/ConsumerGroups/[consumerGroup]/Partitions/[partitionId]
// Endpoint=sb://<eventhub>.servicebus.windows.net/;SharedAccessKeyName=<sharedAccessKeyName>;SharedAccessKey=<sharedAccessKey>;EntityPath=<entity-path>
// wss://<eventhub>.servicebus.windows.net:443/$servicebus/websocket -->
var hostName = "<eventhub>.servicebus.windows.net";
var sharedAccessKeyName = "RootManageSharedAccessKey";
var sharedAccessKey = "<sharedAccessKey>";
var wsServer = "wss://" + hostName + ":443/$servicebus/websocket"; // Our websocket server
var eventhubName = "<eventhubName>";
var eventHubConsumerGroup = "<eventhubConsumerGroup>";
var connectionSettings = {
"hostname": hostName,
"container_id": "conn" + new Date().getTime(),
"max_frame_size": 4294967295,
"channel_max": 65535,
"idle_timeout": 120000,
"outgoing_locales": 'en-US',
"incoming_locales": 'en-US',
"offered_capabilities": null,
"desired_capabilities": null,
"properties": {},
"connection_details": null, // Will be set below!
"reconnect": false,
"username": sharedAccessKeyName,
"password": sharedAccessKey,
"onSuccess": null,
"onFailure": null,
}
// Connect to the EventHub over AMQP
var sender;
var client = require("rhea");
var ws = client.websocket_connect(WebSocket);
connectionSettings.connection_details = ws(wsServer, ["AMQPWSB10"]);
client.on('connection_open', function (ctx) {
console.log('Connection Opened');
// Connect to a topic, $management contains our partitions
// More: https://github.com/Azure/azure-event-hubs-node/blob/91ba72d47f0fbc0e07318c221102bbcb01df271a/send_receive/lib/client.js#L169
connection.open_receiver('$management');
sender = connection.open_sender('$management');
});
client.on('connection_error', function (ctx) {
console.log('Connection Error: ' + ctx);
});
client.on('connection_close', function (ctx) {
console.log('Connection Closed');
});
client.on('receiver_open', function (ctx) {
console.log('Receiver open');
console.log(ctx);
});
client.on('sendable', function (context) {
// Our sender to the $management topic has been opened
// Send a message to our $management topic to fetch our partitions
sender.send({
body: client.message.data_section(str2ab('[]')),
application_properties: {
operation: 'READ',
name: eventhubName,
type: 'com.microsoft:eventhub'
}
});
});
client.on("message", function (context) {
if (context.receiver.source.address === '$management') {
var p = context.message.body;
var partitionCount = p.partition_count;
// Open receivers for all my partitions
for (var i = 0; i < partitionCount; i++) {
console.log('Opening receiver for ' + '/' + eventhubName + '/ConsumerGroups/' + eventHubConsumerGroup + '/Partitions/' + i)
connection.open_receiver({
source: {
address: '/' + eventhubName + '/ConsumerGroups/' + eventHubConsumerGroup + '/Partitions/' + i,
filter: client.filter.selector("amqp.annotation.x-opt-enqueuedtimeutc > " + (new Date().getTime()))
}
});
}
}
// Process message
if (!context.message.body.content) {
return;
}
var decodedMessage = Utf8ArrayToStr(context.message.body.content);
var decodedMessages = decodedMessage.split('\n'); // Apparently multiple json messages per payload
console.log(decodedMessages);
});
client.on("error", function (ctx) {
console.log(ctx);
})
var connection = client.connect(connectionSettings);
References
- https://github.com/amqp/rhea
- https://blogs.msdn.microsoft.com/zhiqing/2017/03/28/how-to-use-amqp-protocol-in-browser-javascript/
- https://github.com/michaeljqzq/NOTAZUREIotDevTool_Service/blob/master/util.ts
- https://github.com/Azure/azure-event-hubs-node/blob/91ba72d47f0fbc0e07318c221102bbcb01df271a/send_receive/lib/client.js#L169