Merge pull request #34 from dolittle/fix-api

Fix public api

Author: joelhoisko

Samples/EventHorizon/index.ts

@@ -9,7 +9,7 @@ import { PartitionId } from '@dolittle/sdk.events';
 
 const client = Client
     .forMicroservice('a14bb24e-51f3-4d83-9eba-44c4cffe6bb9')
-    .connectToRuntime('localhost', 50055)
+    .withRuntimeOn('localhost', 50055)
     .withEventTypes(eventTypes =>
         eventTypes.register(MyEvent))
     .withEventHandlers(eventHandlers =>

Source/artifacts/EventTypesBuilder.ts

@@ -3,6 +3,7 @@
 
 import { Constructor } from '@dolittle/types';
 import { Guid } from '@dolittle/rudiments';
+
 import { Generation } from './Generation';
 import { EventType } from './EventType';
 import { IEventTypes } from './IEventTypes';

Source/eventHorizon/SubscriptionBuilderForConsumerScope.ts

@@ -4,20 +4,21 @@
 
 import { StreamId, PartitionId, ScopeId } from '@dolittle/sdk.events';
 import { MicroserviceId, TenantId } from '@dolittle/sdk.execution';
+import { Observable } from 'rxjs';
+import { filter } from 'rxjs/operators';
 
 import { Subscription } from './Subscription';
-import { SubscriptionCallbacks, SubscriptionCompleted, SubscriptionFailed, SubscriptionSucceeded } from './SubscriptionCallbacks';
+import { SubscriptionCallbackArguments, SubscriptionCallbacks, SubscriptionCompleted, SubscriptionFailed, SubscriptionSucceeded } from './SubscriptionCallbacks';
 
 /**
  * Represents the builder for building subscriptions on a tenant.
  */
 export class SubscriptionBuilderForConsumerScope {
+    private readonly _callbacks: ((subscriptionCallback: SubscriptionCallbacks) => void)[] = [];
 
-    private readonly _callbacks: SubscriptionCallbacks = new SubscriptionCallbacks();
     /**
      * Initializes a new instance of {@link SubscriptionBuilderForConsumerScope}.
      * @param {MicroserviceId} _producerMicroserviceId The microservice the subscriptions are for.
-     * @param {Observable<SubscriptionCallbackArguments>} responsesSource The source of responses.
      */
     constructor(
         private readonly _producerMicroserviceId: MicroserviceId,
@@ -34,7 +35,7 @@ export class SubscriptionBuilderForConsumerScope {
      * @summary The callback will be called on each subscription.
      */
     onCompleted(completed: SubscriptionCompleted): SubscriptionBuilderForConsumerScope {
-        this._callbacks.onCompleted(completed);
+        this._callbacks.push(subscriptionCallbacks => subscriptionCallbacks.onCompleted(completed));
         return this;
     }
 
@@ -45,7 +46,7 @@ export class SubscriptionBuilderForConsumerScope {
      * @summary The callback will be called on each subscription.
      */
     onSuccess(succeeded: SubscriptionSucceeded): SubscriptionBuilderForConsumerScope {
-        this._callbacks.onSucceeded(succeeded);
+        this._callbacks.push(subscriptionCallbacks => subscriptionCallbacks.onSucceeded(succeeded));
         return this;
     }
 
@@ -56,7 +57,7 @@ export class SubscriptionBuilderForConsumerScope {
      * @summary The callback will be called on each subscription.
      */
     onFailure(failed: SubscriptionFailed): SubscriptionBuilderForConsumerScope {
-        this._callbacks.onFailed(failed);
+        this._callbacks.push(subscriptionCallbacks => subscriptionCallbacks.onFailed(failed));
         return this;
     }
 
@@ -65,7 +66,23 @@ export class SubscriptionBuilderForConsumerScope {
      * @param {Observable<SubscriptionCallbackArguments} callbackArgumentsSource The observable source of responses.
      * @returns {Subscription}
      */
-    build(): Subscription {
-        return new Subscription(this._consumerScopeId, this._producerMicroserviceId, this._producerTenantId, this._producerStreamId, this._producerPartitionId, this._callbacks);
+    build(callbackArgumentsSource: Observable<SubscriptionCallbackArguments>): Subscription {
+        const subscriptionCallbacks = new SubscriptionCallbacks(
+            callbackArgumentsSource.pipe(filter(_ =>
+                _.subscription.microservice.equals(this._producerMicroserviceId) &&
+                _.subscription.partition.equals(this._producerPartitionId) &&
+                _.subscription.scope.equals(this._consumerScopeId) &&
+                _.subscription.stream.equals(this._producerStreamId) &&
+                _.subscription.tenant.equals(this._producerTenantId))));
+        for (const callback of this._callbacks) {
+            callback(subscriptionCallbacks);
+        }
+        return new Subscription(
+            this._consumerScopeId,
+            this._producerMicroserviceId,
+            this._producerTenantId,
+            this._producerStreamId,
+            this._producerPartitionId,
+            subscriptionCallbacks);
     }
 }

Source/eventHorizon/SubscriptionBuilderForProducerMicroservice.ts

@@ -9,14 +9,17 @@ import { Subscription } from './Subscription';
 import { SubscriptionBuilderForProducerTenant  } from './SubscriptionBuilderForProducerTenant';
 import { SubscriptionDefinitionIncomplete } from './SubscriptionDefinitionIncomplete';
 import { SubscriptionBuilderMethodAlreadyCalled } from './SubscriptionBuilderMethodAlreadyCalled';
+import { SubscriptionCallbackArguments } from './SubscriptionCallbacks';
+import { Observable } from 'rxjs';
 
 /**
  * Represents the builder for building subscriptions on a tenant.
  */
 export class SubscriptionBuilderForProducerMicroservice {
-    private _producerTenant?: TenantId;
+    private _producerTenantId?: TenantId;
     private _builder?: SubscriptionBuilderForProducerTenant;
 
+
     /**
      * Initializes a new instance of {@link SubscriptionBuilderForProducerMicroservice}.
      * @param {MicroserviceId} _producerMicroserviceId The microservice the subscriptions are for.
@@ -27,12 +30,12 @@ export class SubscriptionBuilderForProducerMicroservice {
 
     /**
      * Specifies from which tenant we should get events from in the other microservice.
-     * @param {Guid | string} tenant Tenant for the subscription.
+     * @param {Guid | string} tenantId Tenant for the subscription.
      */
-    fromProducerTenant(tenant: Guid | string): SubscriptionBuilderForProducerTenant {
+    fromProducerTenant(tenantId: Guid | string): SubscriptionBuilderForProducerTenant {
         this.throwIfProducerTenantIsAlreadyDefined();
-        this._producerTenant = TenantId.from(tenant);
-        this._builder = new SubscriptionBuilderForProducerTenant(this._producerMicroserviceId, this._producerTenant);
+        this._producerTenantId = TenantId.from(tenantId);
+        this._builder = new SubscriptionBuilderForProducerTenant(this._producerMicroserviceId, this._producerTenantId);
         return this._builder;
     }
 
@@ -41,18 +44,18 @@ export class SubscriptionBuilderForProducerMicroservice {
      * @param {Observable<SubscriptionCallbackArguments} callbackArgumentsSource The observable source of responses.
      * @returns {Subscription}
      */
-    build(): Subscription {
+    build(callbackArgumentsSource: Observable<SubscriptionCallbackArguments>): Subscription {
         this.throwIfProducerTenantIsNotDefined();
-        return this._builder!.build();
+        return this._builder!.build(callbackArgumentsSource);
     }
 
     private throwIfProducerTenantIsAlreadyDefined() {
-        if (this._producerTenant) {
+        if (this._producerTenantId) {
             throw new SubscriptionBuilderMethodAlreadyCalled('fromProducerTenant()');
         }
     }
     private throwIfProducerTenantIsNotDefined() {
-        if (!this._producerTenant) {
+        if (!this._producerTenantId) {
             throw new SubscriptionDefinitionIncomplete('Producer Tenant', 'Call fromProducerTenant()');
         }
     }

Source/eventHorizon/SubscriptionBuilderForProducerPartition.ts

@@ -10,12 +10,14 @@ import { Subscription } from './Subscription';
 import { SubscriptionBuilderMethodAlreadyCalled } from './SubscriptionBuilderMethodAlreadyCalled';
 import { SubscriptionDefinitionIncomplete } from './SubscriptionDefinitionIncomplete';
 import { SubscriptionBuilderForConsumerScope } from './SubscriptionBuilderForConsumerScope';
+import { SubscriptionCallbackArguments } from './SubscriptionCallbacks';
+import { Observable } from 'rxjs';
 
 /**
  * Represents the builder for building subscriptions on a tenant.
  */
 export class SubscriptionBuilderForProducerPartition {
-    private _consumerScope?: ScopeId;
+    private _consumerScopeId?: ScopeId;
     private _builder?: SubscriptionBuilderForConsumerScope;
 
     /**
@@ -32,12 +34,17 @@ export class SubscriptionBuilderForProducerPartition {
 
     /**
      * Sets the producer stream to subscribe to events from.
-     * @param {Guid | string} consumerScope Stream to subscribe to events from.
+     * @param {Guid | string} scopeId Stream to subscribe to events from.
      */
-    toScope(consumerScope: Guid | string): SubscriptionBuilderForConsumerScope {
+    toScope(scopeId: Guid | string): SubscriptionBuilderForConsumerScope {
         this.throwIfConsumerScopeIsAlreadyDefined();
-        this._consumerScope = ScopeId.from(consumerScope);
-        this._builder = new SubscriptionBuilderForConsumerScope(this._producerMicroserviceId, this._producerTenantId, this._producerStreamId, this._producerPartitionId, this._consumerScope);
+        this._consumerScopeId = ScopeId.from(scopeId);
+        this._builder = new SubscriptionBuilderForConsumerScope(
+            this._producerMicroserviceId,
+            this._producerTenantId,
+            this._producerStreamId,
+            this._producerPartitionId,
+            this._consumerScopeId);
         return this._builder;
     }
 
@@ -46,18 +53,18 @@ export class SubscriptionBuilderForProducerPartition {
      * @param {Observable<SubscriptionCallbackArguments} callbackArgumentsSource The observable source of responses.
      * @returns {Subscription}
      */
-    build(): Subscription {
+    build(callbackArgumentsSource: Observable<SubscriptionCallbackArguments>): Subscription {
         this.throwIfConsumerScopeIsNotDefined();
-        return this._builder!.build();
+        return this._builder!.build(callbackArgumentsSource);
     }
 
     private throwIfConsumerScopeIsAlreadyDefined() {
-        if (this._consumerScope) {
+        if (this._consumerScopeId) {
             throw new SubscriptionBuilderMethodAlreadyCalled('toScope()');
         }
     }
     private throwIfConsumerScopeIsNotDefined() {
-        if (!this._consumerScope) {
+        if (!this._consumerScopeId) {
             throw new SubscriptionDefinitionIncomplete('Scope', 'Call toScope() with a non-default scope');
         }
     }

Source/eventHorizon/SubscriptionBuilderForProducerStream.ts

@@ -10,6 +10,8 @@ import { Subscription } from './Subscription';
 import { SubscriptionBuilderMethodAlreadyCalled } from './SubscriptionBuilderMethodAlreadyCalled';
 import { SubscriptionDefinitionIncomplete } from './SubscriptionDefinitionIncomplete';
 import { SubscriptionBuilderForProducerPartition } from './SubscriptionBuilderForProducerPartition';
+import { Observable } from 'rxjs';
+import { SubscriptionCallbackArguments } from './SubscriptionCallbacks';
 
 /**
  * Represents the builder for building subscriptions on a tenant.
@@ -36,7 +38,11 @@ export class SubscriptionBuilderForProducerStream {
     fromProducerPartition(partitionId: Guid | string): SubscriptionBuilderForProducerPartition {
         this.throwIfProducerPartitionIsAlreadyDefined();
         this._producerPartitionId = PartitionId.from(partitionId);
-        this._builder = new SubscriptionBuilderForProducerPartition(this._producerMicroserviceId, this._producerTenantId, this._producerStreamId, this._producerPartitionId);
+        this._builder = new SubscriptionBuilderForProducerPartition(
+            this._producerMicroserviceId,
+            this._producerTenantId,
+            this._producerStreamId,
+            this._producerPartitionId);
         return this._builder;
     }
 
@@ -45,9 +51,9 @@ export class SubscriptionBuilderForProducerStream {
      * @param {Observable<SubscriptionCallbackArguments} callbackArgumentsSource The observable source of responses.
      * @returns {Subscription}
      */
-    build(): Subscription {
+    build(callbackArgumentsSource: Observable<SubscriptionCallbackArguments>): Subscription {
         this.throwIfProducerPartitionIsNotDefined();
-        return this._builder!.build();
+        return this._builder!.build(callbackArgumentsSource);
     }
 
     private throwIfProducerPartitionIsAlreadyDefined() {

Source/eventHorizon/SubscriptionBuilderForProducerTenant.ts

@@ -10,6 +10,8 @@ import { Subscription } from './Subscription';
 import { SubscriptionBuilderMethodAlreadyCalled } from './SubscriptionBuilderMethodAlreadyCalled';
 import { SubscriptionDefinitionIncomplete } from './SubscriptionDefinitionIncomplete';
 import { SubscriptionBuilderForProducerStream } from './SubscriptionBuilderForProducerStream';
+import { Observable } from 'rxjs';
+import { SubscriptionCallbackArguments} from './SubscriptionCallbacks';
 
 /**
  * Represents the builder for building subscriptions on a tenant.
@@ -35,7 +37,10 @@ export class SubscriptionBuilderForProducerTenant {
     fromProducerStream(streamId: Guid | string): SubscriptionBuilderForProducerStream {
         this.throwIfProducerStreamIsAlreadyDefined();
         this._producerStreamId = StreamId.from(streamId);
-        this._builder = new SubscriptionBuilderForProducerStream(this._producerMicroserviceId, this._producerTenantId, this._producerStreamId);
+        this._builder = new SubscriptionBuilderForProducerStream(
+            this._producerMicroserviceId,
+            this._producerTenantId,
+            this._producerStreamId);
         return this._builder;
     }
 
@@ -44,9 +49,9 @@ export class SubscriptionBuilderForProducerTenant {
      * @param {Observable<SubscriptionCallbackArguments} callbackArgumentsSource The observable source of responses.
      * @returns {Subscription}
      */
-    build(): Subscription {
+    build(callbackArgumentsSource: Observable<SubscriptionCallbackArguments>): Subscription {
         this.throwIfProducerStreamIsNotDefined();
-        return this._builder!.build();
+        return this._builder!.build(callbackArgumentsSource);
     }
 
     private throwIfProducerStreamIsAlreadyDefined() {

Source/eventHorizon/SubscriptionsBuilder.ts

@@ -25,13 +25,13 @@ export class SubscriptionsBuilder {
 
     /**
      * Configure subscriptions for a tenant in our microservice.
-     * @param {Guid | string} tenant The tenant in our microservice.
+     * @param {Guid | string} tenantId The tenant in our microservice.
      * @param {TenantWithSubscriptionsBuilderCallback} callback The subscriptions builder callback.
      * @returns {SubscriptionsBuilder}
      * @summary Two microservices does not need to be aligned on tenancy. This allows for that purpose.
      */
-    forTenant(tenant: Guid | string, callback: (builder: SubscriptionsBuilderForConsumerTenant) => void): SubscriptionsBuilder {
-        const builder = new SubscriptionsBuilderForConsumerTenant(TenantId.from(tenant), this._callbacks.responses);
+    forTenant(tenantId: Guid | string, callback: (builder: SubscriptionsBuilderForConsumerTenant) => void): SubscriptionsBuilder {
+        const builder = new SubscriptionsBuilderForConsumerTenant(TenantId.from(tenantId), this._callbacks.responses);
         callback(builder);
         this._tenantSubscriptionsBuilders.push(builder);
         return this;

Source/eventHorizon/SubscriptionsBuilderForConsumerTenant.ts

@@ -13,7 +13,7 @@ import { Guid } from '@dolittle/rudiments';
  * Represents the builder of {@link TenantSubscriptions}.
  */
 export class SubscriptionsBuilderForConsumerTenant {
-    private readonly _callbacks: SubscriptionCallbacks = new SubscriptionCallbacks();
+    private readonly _callbacks: SubscriptionCallbacks;
     readonly _subscriptionBuilders: SubscriptionBuilderForProducerMicroservice[] = [];
 
     /**
@@ -22,16 +22,18 @@ export class SubscriptionsBuilderForConsumerTenant {
      * @param {Observable<SubscriptionCallbackArguments>} responsesSource The source of responses.
      */
     constructor(private _consumerTenantId: TenantId, responsesSource: Observable<SubscriptionCallbackArguments>) {
-        this._callbacks = new SubscriptionCallbacks(responsesSource.pipe(filter(_ => _.consumerTenant.toString() === _consumerTenantId.toString())));
+        this._callbacks = new SubscriptionCallbacks(
+            responsesSource.pipe(filter(_ =>
+                _.consumerTenant.toString() === _consumerTenantId.toString())));
     }
 
     /**
      * Sets the producer microservice to subscribe to events from.
-     * @param {Guid | string} microservice Microservice to build for.
+     * @param {Guid | string} microserviceId Microservice to build for.
      * @returns {SubscriptionBuilderForProducerMicroservice}
      */
-    fromProducerMicroservice(microservice: Guid | string): SubscriptionBuilderForProducerMicroservice {
-        const builder = new SubscriptionBuilderForProducerMicroservice(MicroserviceId.from(microservice));
+    fromProducerMicroservice(microserviceId: Guid | string): SubscriptionBuilderForProducerMicroservice {
+        const builder = new SubscriptionBuilderForProducerMicroservice(MicroserviceId.from(microserviceId));
         this._subscriptionBuilders.push(builder);
         return builder;
     }
@@ -74,7 +76,7 @@ export class SubscriptionsBuilderForConsumerTenant {
      * @returns {TenantWithSubscriptions}
      */
     build(): TenantWithSubscriptions {
-        const subscriptions = this._subscriptionBuilders.map(_ => _.build());
+        const subscriptions = this._subscriptionBuilders.map(_ => _.build(this._callbacks.responses));
         return new TenantWithSubscriptions(
             this._consumerTenantId,
             subscriptions,

Source/events.handling/Builder/EventHandlersBuilder.ts

@@ -32,8 +32,7 @@ export class EventHandlersBuilder {
      * @param {EventHandlerBuilderCallback} callback Callback for building out the event handler.
      */
     createEventHandler(eventHandlerId: Guid | string, callback: EventHandlerBuilderCallback): EventHandlersBuilder {
-        const id = EventHandlerId.from(eventHandlerId);
-        const builder = new EventHandlerBuilder(id);
+        const builder = new EventHandlerBuilder(EventHandlerId.from(eventHandlerId));
         callback(builder);
         this._eventHandlerBuilders.push(builder);
         return this;