From bc66e4bc3c9e5792245171bf9638bd2227cd2d77 Mon Sep 17 00:00:00 2001 From: Robert on Asus 305 Date: Tue, 1 Nov 2016 07:26:27 -0700 Subject: [PATCH] Samples: Subscriptions. Subscribe to all events & subscribe to all events, catching up from the beginning. --- .gitignore | 3 ++ README.md | 20 +++++++- samples/{storeevent.js => store-event.js} | 0 .../{subscribe.js => subscribe-all-events.js} | 20 ++++---- samples/subscribe-catchup-all-events.js | 51 +++++++++++++++++++ 5 files changed, 82 insertions(+), 12 deletions(-) rename samples/{storeevent.js => store-event.js} (100%) rename samples/{subscribe.js => subscribe-all-events.js} (74%) create mode 100644 samples/subscribe-catchup-all-events.js diff --git a/.gitignore b/.gitignore index a048be2..0ee2c9b 100644 --- a/.gitignore +++ b/.gitignore @@ -37,3 +37,6 @@ node_modules .vscode lib/ +.eslintrc.json +jsconfig.json + diff --git a/README.md b/README.md index be9e115..130d4e0 100644 --- a/README.md +++ b/README.md @@ -69,9 +69,21 @@ npm install eventstore-node node app.js ``` -## Porting .Net Task to Node.js +### Example: Subscribing to events -.Net Task have been replace with Promise. When executing an async command, i.e. appendToStream you can use then/catch to wait for result/error. +```cd samples``` + +To subscribe to all events from now on (includes example of a filter which ignores events which we aren't interested in): + +```node subscribe-all-events.js``` + +To catch up on all events ever and subscribe to all new ones from now on: + +```node subscribe-catchup-all-events.js``` + +To generate a test event, open a separate console and run: + +```node store-event.js``` ## Running the tests @@ -84,6 +96,10 @@ To execute the tests suites simply run test with npm npm test +## Porting .Net Task to Node.js + +.Net Task have been replace with Promise. When executing an async command, i.e. appendToStream you can use then/catch to wait for result/error. + ## License Ported code is released under the MIT license, see [LICENSE](https://github.com/nicdex/eventstore-node/blob/master/LICENSE). diff --git a/samples/storeevent.js b/samples/store-event.js similarity index 100% rename from samples/storeevent.js rename to samples/store-event.js diff --git a/samples/subscribe.js b/samples/subscribe-all-events.js similarity index 74% rename from samples/subscribe.js rename to samples/subscribe-all-events.js index 8ae678a..fd235ee 100644 --- a/samples/subscribe.js +++ b/samples/subscribe-all-events.js @@ -1,3 +1,5 @@ +// Subscribe to all new events on the $all stream. Filter out any which aren't about "user" aggregates. + var esClient = require('../src/client'); // When running in 'eventstore-node/samples' folder. // var esClient = require('eventstore-node'); // Otherwise @@ -8,6 +10,10 @@ var esConnection = esClient.createConnection({}, {"hostname": "localhost", "port esConnection.connect(); esConnection.once('connected', function (tcpEndPoint) { console.log('Connected to eventstore at ' + tcpEndPoint.hostname + ":" + tcpEndPoint.port); + esConnection.subscribeToAll(resolveLinkTos, eventAppeared, subscriptionDropped, credentialsForAllEventsStream) + .then(function(subscription) { + console.log("subscription.isSubscribedToAll: " + subscription.isSubscribedToAll); + }); }); function belongsToAUserAggregate(event) { @@ -20,28 +26,22 @@ function eventAppeared(subscription, event) { var aggregateId = event.originalEvent.eventStreamId; var eventId = event.originalEvent.eventId; var eventType = event.originalEvent.eventType; - var eventCreated = event.originalEvent.created; - console.log(aggregateId, eventType, eventId, eventCreated); + console.log(aggregateId, eventType, eventId); console.log(event.originalEvent.data.toString() + "\n"); } } function subscriptionDropped(subscription, reason, error) { - if (error) return test.done(error); + if (error) { + console.log(error); + } console.log('Subscription dropped.'); } -esConnection.subscribeToAll(resolveLinkTos, eventAppeared, subscriptionDropped, credentialsForAllEventsStream) - .then(function(subscription) { - console.log("subscription.isSubscribedToAll: " + subscription.isSubscribedToAll); - }) - .catch(console.log("Caught.")); - esConnection.on('error', function (err) { console.log('Error occurred on connection:', err); }); esConnection.on('closed', function (reason) { console.log('Connection closed, reason:', reason); - process.exit(-1); }); diff --git a/samples/subscribe-catchup-all-events.js b/samples/subscribe-catchup-all-events.js new file mode 100644 index 0000000..dac2e30 --- /dev/null +++ b/samples/subscribe-catchup-all-events.js @@ -0,0 +1,51 @@ +// Subscribe to all events on the $all stream. Catch up from the beginning, then listen for any new events as they occur. +// This can be used (e.g.) for subscribers which populate read models. + +var esClient = require('../src/client'); // When running in 'eventstore-node/samples' folder. +// var esClient = require('eventstore-node'); // Otherwise + +const credentialsForAllEventsStream = new esClient.UserCredentials("admin", "changeit"); + +var esConnection = esClient.createConnection({}, {"hostname": "localhost", "port": 1113}); +esConnection.connect(); +esConnection.once('connected', function (tcpEndPoint) { + console.log('Connected to eventstore at ' + tcpEndPoint.hostname + ":" + tcpEndPoint.port); + var subscription = esConnection.subscribeToAllFrom(null, true, eventAppeared, liveProcessingStarted, subscriptionDropped, credentialsForAllEventsStream); + console.log("subscription.isSubscribedToAll: " + subscription.isSubscribedToAll); +}); + +function eventAppeared(subscription, event) { + // This is where to filter out events which the subscriber isn't interested in. + // For an example, see 'subscribe-all-events.js'. + console.log(event.originalEvent.eventStreamId); +} + +function subscriptionDropped(subscription, reason, error) { + if (error) { + console.log(error); + } + console.log('Subscription dropped.'); +} + +function liveProcessingStarted() { + console.log("Caught up with previously stored events. Listening for new events."); + console.log("(To generate a test event, try running 'node store-event.js' in a separate console.)") +} + +function eventAppeared(stream, event) { + console.log(event.originalEvent.eventStreamId, event.originalEvent.eventId, event.originalEvent.eventType); + // Data: + // console.log(event.originalEvent.data.toString()); + + // Position in the event stream. Can be persisted and used to catch up with missed events when re-starting subscribers instead of re-reading + // all events from the beginning. + // console.log(event.originalPosition); +} + +esConnection.on('error', function (err) { + console.log('Error occurred on connection:', err); +}); + +esConnection.on('closed', function (reason) { + console.log('Connection closed, reason:', reason); +});