[Feature] connect to persistent subscription

index.d.ts

@@ -5,7 +5,8 @@ import {
 	DeleteResult as TCPDeleteResult,
 	LiveProcessingStartedCallback,
 	SubscriptionDroppedCallback,
-	ConnectionSettings
+	ConnectionSettings,
+	EventStorePersistentSubscription
 }  from "node-eventstore-client";
 
 import {
@@ -218,6 +219,7 @@ export class TCPClient {
 	};
 	subscribeToStream(streamName: string, onEventAppeared?: MappedEventAppearedCallback<EventStoreSubscription>, onDropped?: SubscriptionDroppedCallback<EventStoreSubscription>, resolveLinkTos?: boolean): Promise<EventStoreSubscription>;
 	subscribeToStreamFrom(streamName: string, fromEventNumber?: number, onEventAppeared?: MappedEventAppearedCallback<EventStoreCatchUpSubscription>, onLiveProcessingStarted?: LiveProcessingStartedCallback, onDropped?: SubscriptionDroppedCallback<EventStoreCatchUpSubscription>, settings?: SubscribeToStreamFromSettings): Promise<EventStoreCatchUpSubscription>;
+	connectToPersistentSubscription(streamName: string, groupName: string, onEventAppeared?: MappedEventAppearedCallback<EventStoreCatchUpSubscription>, onDropped?: SubscriptionDroppedCallback<EventStoreSubscription>): Promise<EventStorePersistentSubscription>;
 	close(): Promise<void>;
 	getPool(): Promise<TCPPool<object>>;
 	closeAllPools(): Promise<void>;

lib/tcpClient/connectToPersistentSubscription.js

@@ -0,0 +1,43 @@
+import connectionManager from './connectionManager';
+import mapEvents from './utilities/mapEvents';
+import client from 'node-eventstore-client';
+import debugModule from 'debug';
+import assert from 'assert';
+
+const debug = debugModule('geteventstore:connectToPersistentSubscription');
+const baseErr = 'Connect to Persistent Subscription - ';
+
+export default (config) => (
+	streamName,
+	groupName,
+	onEventAppeared,
+	onDropped,
+) => new Promise(async (resolve, reject) => {
+	assert(streamName, `${baseErr}Stream Name not provided`);
+	assert(groupName, `${baseErr}Group Name not provided`);
+
+	let connection;
+	const onEvent = (sub, ev) => {
+		const mappedEvent = mapEvents([ev])[0];
+		if (mappedEvent) onEventAppeared(sub, mappedEvent);
+	};
+
+	const onConnected = async () => {
+		try {
+			const subscription = await connection.connectToPersistentSubscription(
+				streamName, groupName, onEvent, onDropped, new client.UserCredentials(config.credentials.username, config.credentials.password), 10, true
+			);
+
+			debug('', 'Persistent Subscription: %j', subscription);
+			resolve(subscription);
+		} catch (ex) {
+			reject(ex);
+		}
+	};
+
+	try {
+		connection = await connectionManager.create(config, onConnected, true);
+	} catch (ex) {
+		reject(ex);
+	}
+});

lib/tcpClient/index.js

@@ -1,3 +1,4 @@
+import connectToPersistentSubscription from './connectToPersistentSubscription';
 import subscribeToStreamFrom from './subscribeToStreamFrom';
 import getAllStreamEvents from './getAllStreamEvents';
 import checkStreamExists from './checkStreamExists';
@@ -58,6 +59,7 @@ export default class TCPClient {
 		this.eventEnumerator = eventEnumerator(_config);
 		this.subscribeToStream = subscribeToStream(_config);
 		this.subscribeToStreamFrom = subscribeToStreamFrom(_config);
+		this.connectToPersistentSubscription = connectToPersistentSubscription(_config);
 		this.close = connectionManager.close(_config);
 		this.getPool = connectionManager.getPool(_config);
 		this.closeAllPools = connectionManager.closeAllPools;