Azure
/
azure-event-hubs-node
зеркало из
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность

move spike code

This commit is contained in:
Damon Barry 2015-11-17 10:08:00 -08:00
Родитель fa43894735
Коммит 264e8bf6bb
14 изменённых файлов: 0 добавлений и 558 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

1
.jshintignore
Просмотреть файл

@ -1 +0,0 @@
**/node_modules

20
.jshintrc
Просмотреть файл

@ -1,20 +0,0 @@
{
"node": true,
"mocha": true,
"bitwise": true,
"curly": false,
"eqeqeq": true,
"forin": true,
"immed": true,
"latedef": false,
"newcap": true,
"noarg": true,
"noempty": true,
"nonew": true,
"regexp": true,
"undef": true,
"unused": true,
"strict": true,
"globalstrict": true,
"trailing": true
}

74
build/build.cmd
Просмотреть файл

@ -1,74 +0,0 @@
@REM Copyright (c) Microsoft. All rights reserved.
@REM Licensed under the MIT license. See LICENSE file in the project root for full license information.
@setlocal
@echo off
set node-root=%~dp0..
rem // resolve to fully qualified path
for %%i in ("%node-root%") do set node-root=%%~fi
rem ---------------------------------------------------------------------------
rem -- parse script arguments
rem ---------------------------------------------------------------------------
set min-output=0
set integration-tests=0
:args-loop
if "%1" equ "" goto args-done
if "%1" equ "--min" goto arg-min-output
if "%1" equ "--integration-tests" goto arg-integration-tests
call :usage && exit /b 1
:arg-min-output
set min-output=1
goto args-continue
:arg-integration-tests
set integration-tests=1
goto args-continue
:args-continue
shift
goto args-loop
:args-done
if %min-output%==0 if %integration-tests%==0 set "npm-command=npm -s test"
if %min-output%==0 if %integration-tests%==1 set "npm-command=npm -s run lint && npm -s run alltest"
if %min-output%==1 if %integration-tests%==0 set "npm-command=npm -s run lint && npm -s run unittest-min"
if %min-output%==1 if %integration-tests%==1 set "npm-command=npm -s run ci"
rem ---------------------------------------------------------------------------
rem -- lint and run tests
rem ---------------------------------------------------------------------------
echo.
if %integration-tests%==0 echo -- Linting and running unit tests --
if %integration-tests%==1 echo -- Linting and running unit + integration tests --
echo.
call :lint-and-test %node-root%\
if errorlevel 1 goto :eof
goto :eof
rem ---------------------------------------------------------------------------
rem -- helper subroutines
rem ---------------------------------------------------------------------------
:usage
echo Lint code and run tests.
echo build.cmd [options]
echo options:
echo --min minimize display output
echo --integration-tests run integration tests too (unit tests always run)
goto :eof
:lint-and-test
cd "%1"
echo %cd%
call %npm-command%
goto :eof

65
build/build.sh
Просмотреть файл

@ -1,65 +0,0 @@
#!/bin/sh
# Copyright (c) Microsoft. All rights reserved.
# Licensed under the MIT license. See LICENSE file in the project root for full license information.
min_output=
integration_tests=
npm_command=
node_root=$(cd "$(dirname "$0")/.." && pwd)
cd $node_root
usage ()
{
echo "Lint code and run tests."
echo "build.sh [options]"
echo "options"
echo " --min minimize display output"
echo " --integration-tests run integration tests too (unit tests always run)"
exit 1
}
process_args ()
{
min_output=0
integration_tests=0
for arg in $*
do
case "$arg" in
"--min" ) min_output=1;;
"--integration-tests" ) integration_tests=1;;
* ) usage;;
esac
done
case "$min_output$integration_tests" in
"00" ) npm_command="npm -s test";;
"01" ) npm_command="npm -s run lint && npm -s run alltest";;
"10" ) npm_command="npm -s run lint && npm -s run unittest-min";;
"11" ) npm_command="npm -s run ci";;
esac
}
lint_and_test ()
{
cd "$1"
pwd
eval $npm_command
}
process_args $*
echo ""
if [ $integration_tests -eq 0 ]
then
echo "-- Linting and running unit tests --"
else
echo "-- Linting and running unit + integration tests --"
fi
echo ""
lint_and_test $node_root
[ $? -eq 0 ] || exit $?

12
build/dev-setup.cmd
Просмотреть файл

@ -1,12 +0,0 @@
@REM Copyright (c) Microsoft. All rights reserved.
@REM Licensed under the MIT license. See LICENSE file in the project root for full license information.
@setlocal
@echo off
set node-root=%~dp0..
REM // resolve to fully qualified path
for %%i in ("%node-root%") do set node-root=%%~fi
cd %node-root%
call npm install

10
build/dev-setup.sh
Просмотреть файл

@ -1,10 +0,0 @@
#!/bin/sh
# Copyright (c) Microsoft. All rights reserved.
# Licensed under the MIT license. See LICENSE file in the project root for full license information.
node_root=$(cd "$(dirname "$0")/.." && pwd)
cd $node_root
npm install
[ $? -eq 0 ] || exit $?

10
index.js
Просмотреть файл

@ -1,10 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
'use strict';
module.exports = {
EventData: require('./lib/eventdata'),
Receiver: require("./lib/eventhubreceiver"),
Client: require("./lib/eventhubclient")
};

19
lib/_eventhubclient_test.js
Просмотреть файл

@ -1,19 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
'use strict';
var assert = require('chai').assert;
var EventHubClient = require('./eventhubclient.js');
var goodConnString = "Endpoint=sb://xxx.servicebus.windows.net/;SharedAccessKeyName=yyy;SharedAccessKey=zzz;EntityPath=uuu";
describe('EventHubClient', function () {
describe('#StartReceive', function () {
it('starts receiving', function () {
var eventhubclient = new EventHubClient(goodConnString);
assert.isNotNull(eventhubclient);
});
});
});

44
lib/eventdata.js
Просмотреть файл

@ -1,44 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
'use strict';
/**
* @class EventData
* @classdesc Constructs a {@linkcode EventData} object.
* @param {String} bytes The event payload as a byte array.
*/
function EventData(bytes, systemProperties) {
Object.defineProperties(this, {
'PartitionKey':{
value: "",
writable: false
},
'Bytes':{
value: bytes,
writable: false
},
'EnqueuedTimeUtc':{
value: null,
writable: false
},
'Offset':{
value: "",
writable: false
},
'Properties':{
value: null,
writable: true
},
'SequenceNumber':{
value: 0,
writable: false
},
'SystemProperties':{
value: systemProperties,
writable: false
}
});
}
module.exports = EventData;

147
lib/eventhubclient.js
Просмотреть файл

@ -1,147 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
'use strict';
var amqp10 = require('amqp10');
var Promise = require('bluebird');
var EventHubData = require('./eventdata');
var EventHubReceiver = require("./eventhubreceiver");
var managementEndpoint = '$management';
function parseEventHubConnString(connString)
{
var config = {
host: '',
eventHubName: '',
keyName: '',
key: ''
};
var configArray = connString.split(';');
configArray.forEach(function(element) {
var res = element.match("Endpoint=sb://([^/]*)");
if (res !== null) {
config.host = res[1];
}
res = element.match("SharedAccessKeyName=(.*)");
if (res !== null) {
config.keyName = res[1];
}
res = element.match("SharedAccessKey=(.*)");
if (res !== null) {
config.key = res[1];
}
res = element.match("EntityPath=(.*)");
if (res !== null) {
config.eventHubName = res[1];
}
});
return config;
}
/**
* @class EventHubClient
* @classdesc Constructs a {@linkcode EventHubClient} object with the given connection string
* @param {String} connString The EventHub connection string
* @param {String} path The EventHub path. Optional.
*/
function EventHubClient(connString, path) {
this.config = parseEventHubConnString(connString);
if (path) {
this.config.eventHubName = path;
}
if (!this.config.eventHubName || this.config.eventHubName.length === 0)
{
throw new Error('No event hub name specified');
}
this.uri = 'amqps://' +
encodeURIComponent(this.config.keyName) + ':' +
encodeURIComponent(this.config.key) + '@' +
this.config.host;
this.amqpClient = new amqp10.Client(amqp10.Policy.EventHub);
}
/**
* The [Send]{@linkcode EventHubClient#Send} method sends one Event to the Event Hub.
*/
EventHubClient.prototype.Send = function(event) {
/* This code has to be reworked so that it uses the same amqpClient, there is no need to have 2 clients and thus 2 connecitons open */
if (typeof(event) !== typeof(EventHubData)) {
throw new Error('Expected an EventHubData');
}
var self = this;
var endpoint = '/' + this.config.eventHubName;
return new Promise(function(resolve, reject) {
self.amqpClient.connect(self.uri).then(function() {
self.amqpClient.createSender(endpoint).then(function(sender) {
sender.on('errorReceived', function (tx_err) {
reject(new Error('error sending request to Event Hub management endpoint: ' + tx_err));
});
var request = { body: event.Bytes, properties: { messageId: '424242' } };
sender.send(request);
resolve();
});
});
});
};
/**
* The [SendBatch]{@linkcode EventHubClient#SendBatch} method sends a batch of Events to the Event Hub.
*/
EventHubClient.prototype.SendBatch = function(events) {
/* Not implemented yet */
throw new Error('Not implemented (trying to send events:' + JSON.stringify(events) + ')');
};
/**
* The [GetPartitionIds]{@linkcode EventHubClient#GetPartitionIds} method gets the partition Ids for an EventHub.
*/
EventHubClient.prototype.GetPartitionIds = function() {
var self = this;
return new Promise(function(resolve, reject) {
var ehName = self.config.eventHubName;
var rxName = 'eventhubclient-rx';
var rxOptions = { attach: { target: { address: rxName } } };
self.amqpClient.connect(self.uri).then(function() {
/* create a sender to send the request to the $management endpoint */
self.amqpClient.createSender(managementEndpoint).then(function(sender) {
sender.on('errorReceived', function (tx_err) {
reject(new Error('error sending request to Event Hub management endpoint: ' + tx_err));
});
var request = { body: "stub", properties: { messageId: '424242', replyTo: rxName }, applicationProperties: { operation: "READ", name: ehName, type: "com.microsoft:eventhub" } };
return sender.send(request);
});
/* create a receiver for the management endpoint to receive the partition count */
self.amqpClient.createReceiver(managementEndpoint, rxOptions).then(function(receiver) {
receiver.on('errorReceived', function (rx_err) {
reject(new Error('error receiving reply from Event Hub management endpoint: ' + rx_err));
});
receiver.on('message', function (msg) {
return resolve(msg.body.partition_ids);
});
});
});
});
};
/**
* The [CreateReceiver]{@linkcode EventHubClient#CreateReceiver} method creates a new
* {@linkcode EventHubReceiver} instance.
* @param {String} consumerGroup The consumer group to use for the new receiver.
* @param {String} partitionId The partition Id to use for the new receiver.
*/
EventHubClient.prototype.CreateReceiver = function(consumerGroup, partitionId) {
return new EventHubReceiver(this.amqpClient, '/' + this.config.eventHubName + '/ConsumerGroups/' + consumerGroup + '/Partitions/' + partitionId);
};
module.exports = EventHubClient;

83
lib/eventhubreceiver.js
Просмотреть файл

@ -1,83 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
'use strict';
var EventData = require("./eventdata");
var Promise = require('bluebird');
var EventEmitter = require('events').EventEmitter;
var util = require('util');
/**
* @class EventHubReceiver
* @classdesc Constructs an {@linkcode EventHubReceiver} object
*/
var EventHubReceiver = function (amqpClient, endpoint) {
this.amqpClient = amqpClient;
this.endpoint = endpoint;
};
util.inherits(EventHubReceiver, EventEmitter);
// On receiver event received
EventHubReceiver.EventReceived = 'eventReceived';
// On receive error
EventHubReceiver.Error = 'error';
/* Notes: StartReceive shall handle retries
* onError shall be emitted after the retries have been exhausted
* EventHubReceiver shall support redirect
*/
/**
* The [StartReceive]{@linkcode EventHubReceiver#StartReceive} method starts
* receiving events from the event hub for the specified partition.
* @param startTime The startTime to use as filter for the events being received.
*/
EventHubReceiver.prototype.StartReceive = function(startTime) {
if (startTime !== null) {
console.log('Start time:' + startTime);
}
var self = this;
return new Promise(function(resolve) {
self.amqpClient.createReceiver(self.endpoint).then(function (amqpReceiver) {
amqpReceiver.on('message', function (message) {
var eventData = new EventData(message.body, message.annotations.value);
self.emit(EventHubReceiver.EventReceived, eventData);
});
amqpReceiver.on('errorReceived', function(rx_err) {
self.emit(EventHubReceiver.Error, rx_err);
});
});
resolve();
});
};
/**
* The [StartReceiveFromOffset]{@linkcode EventHubReceiver#StartReceiveFromOffset} method starts
* receiving events from the event hub, while filtering events starting at a certian offset.
* @param startOffset The start offset to use as filter for the events being received.
*/
EventHubReceiver.prototype.StartReceiveFromOffset = function(startOffset) {
if (startOffset !== null) {
console.log('Start offset:' + startOffset);
}
var self = this;
return new Promise(function(resolve) {
self.amqpClient.createReceiver(self.endpoint).then(function (amqpReceiver) {
amqpReceiver.on('message', function (message) {
var eventData = new EventData(message.body, message.annotations.value);
self.emit(EventHubReceiver.MessageReceived, eventData);
});
amqpReceiver.on('errorReceived', function(rx_err) {
self.emit(EventHubReceiver.Error, rx_err);
});
});
resolve();
});
};
module.exports = EventHubReceiver;

37
package.json
Просмотреть файл

@ -1,37 +0,0 @@
{
"name": "event-hub-client",
"version": "1.0.0-preview.1",
"description": "Azure EventHub client",
"author": "Microsoft Corporation",
"license": "MIT",
"main": "index.js",
"dependencies": {
"amqp10": "^2.0.1",
"bluebird": "^2.10.2"
},
"devDependencies": {
"chai": "^3.4.0",
"jshint": "^2.8.0",
"mocha": "^2.3.3"
},
"engines": {
"node": "^0.12 || ^4"
},
"repository": {
"type": "git",
"url": "git+https://github.com/Azure/azure-event-hubs.git"
},
"bugs": {
"url": "https://github.com/Azure/azure-event-hubs/issues"
},
"homepage": "https://github.com/Azure/azure-event-hubs#readme",
"scripts": {
"lint": "jshint --show-non-errors .",
"unittest-min": "mocha --reporter dot lib/_*_test.js",
"alltest-min": "mocha --reporter dot lib/_*_test*.js",
"unittest": "mocha --reporter spec lib/_*_test.js",
"alltest": "mocha --reporter spec lib/_*_test*.js",
"ci": "npm -s run lint && npm -s run alltest-min",
"test": "npm -s run lint && npm -s run unittest"
}
}

25
samples/receivesample.js
Просмотреть файл

@ -1,25 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
'use strict';
var EventHub = require('../index.js');
var ehClient = new EventHub.Client("Endpoint=sb://xxx.servicebus.windows.net/;SharedAccessKeyName=yyy;SharedAccessKey=zzz;EntityPath=uuu");
ehClient.GetPartitionIds().then(function(partitionIds) {
console.log('PartCount=' + partitionIds.length);
var receiver = ehClient.CreateReceiver("$Default", "0");
/* start receiving */
receiver.StartReceive(Date.Now).then(function() {
receiver.on('error', function(error) {
console.log('Receive error:' + error);
});
receiver.on('eventReceived', function(eventData) {
console.log('Event received: ');
console.log(eventData.Bytes);
console.log(eventData.SystemProperties);
console.log('');
});
});
});

11
samples/sendsample.js
Просмотреть файл

@ -1,11 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
'use strict';
var EventHub = require('../index.js');
var ehClient = new EventHub.Client("Endpoint=sb://xxx.servicebus.windows.net/;SharedAccessKeyName=yyy;SharedAccessKey=zzz;EntityPath=uuu");
var eventData = new EventHub.EventData(new Buffer("test"));
ehClient.Send(eventData).then(function() {
});