feat(rpc/server): emits an error if a call subscription completes bef…

…ore emitting any value

packages/rpc/src/server/RPCServer.ts

@@ -3,7 +3,9 @@ import {
   application,
   CollectionTreeDeclaration,
   services,
-  query
+  query,
+  collections,
+  types
 } from '@karmic/core';
 import { RPCServerOptions, RPCServerConnection } from './types';
 import { createDefaults } from './defaults';
@@ -19,7 +21,10 @@ export class RPCServer {
   ) {
     const opts = Object.assign(createDefaults(), options);
     const app = application(
-      collection,
+      collections(
+        collection,
+        types({ [opts.complete.name]: opts.complete.item })
+      ),
       options && options.fallback
         ? { fallback: options.fallback, children: opts.children }
         : { children: opts.children }
@@ -42,7 +47,7 @@ export class RPCServer {
         ...app.flatten(':')
       },
       opts.parser,
-      createEnsureError(this.declaration)
+      createEnsureError(this.declaration, opts.complete)
     );
   }
   public connect(connection: RPCServerConnection): () => void {

packages/rpc/src/server/ServerManager/ChannelManager.ts

@@ -1,13 +1,13 @@
 import { Subscription, Observable } from 'rxjs';
 import { PublicError } from '@karmic/core';
 import { RPCNotification, RPCErrorResponse, RPCSingleResponse } from '~/types';
-import { getError } from '../errors';
+import { getError, EnsureErrorType, GetErrorType } from '../errors';
 
 export class ChannelManager {
   private active: { [key: string]: boolean };
   private subscriptions: { [key: string]: Subscription };
-  private ensure: (error: Error) => PublicError;
-  public constructor(ensure: (error: Error) => PublicError) {
+  private ensure: (error: EnsureErrorType) => PublicError;
+  public constructor(ensure: (error: EnsureErrorType) => PublicError) {
     this.active = {};
     this.subscriptions = {};
     this.ensure = ensure;
@@ -20,7 +20,7 @@ export class ChannelManager {
   }
   public error(
     id: string | number | null,
-    error: 'ParseError' | 'InvalidRequest' | Error,
+    error: GetErrorType | Error,
     cb: (data: RPCErrorResponse) => void
   ): void {
     cb({
@@ -67,11 +67,13 @@ export class ChannelManager {
   ): void {
     this.setActive(id, true);
 
+    let hasEmitted = false;
     this.setSubscription(
       id,
       source.subscribe({
         next: (data: any) => {
           if (!this.getActive(id)) return;
+          hasEmitted = true;
           cb({
             jsonrpc: '2.0',
             id,
@@ -80,6 +82,7 @@ export class ChannelManager {
         },
         error: (err: Error) => {
           if (!this.getActive(id)) return;
+          hasEmitted = true;
           cb({
             jsonrpc: '2.0',
             id,
@@ -89,11 +92,21 @@ export class ChannelManager {
         },
         complete: () => {
           if (!this.getActive(id)) return;
-          cb({
-            jsonrpc: '2.0',
-            method: ':complete',
-            params: { id }
-          });
+          if (hasEmitted) {
+            cb({
+              jsonrpc: '2.0',
+              method: ':complete',
+              params: { id }
+            });
+          } else {
+            // If no values have been emitted
+            // before :complete, we emit an error
+            cb({
+              jsonrpc: '2.0',
+              id,
+              error: getError(this.ensure('EarlyComplete'))
+            });
+          }
           this.close(id);
         }
       })

packages/rpc/src/server/ServerManager/ServerManager.ts

@@ -4,17 +4,18 @@ import { DataInput, DataParser } from '~/types';
 import { ChannelManager } from './ChannelManager';
 import { resolve } from './resolve';
 import { LazyPromist, until } from 'promist';
+import { EnsureErrorType } from '../errors';
 
 export class ServerManager {
   private id: number;
   private routes: ApplicationServices;
   private parser: DataParser;
-  private ensure: (error: Error) => PublicError;
+  private ensure: (error: EnsureErrorType) => PublicError;
   private disconnects: { [key: string]: () => void };
   public constructor(
     routes: ApplicationServices,
     parser: DataParser,
-    ensure: (error: Error) => PublicError
+    ensure: (error: EnsureErrorType) => PublicError
   ) {
     this.id = 0;
     this.routes = routes;

packages/rpc/src/server/defaults.ts

@@ -1,9 +1,14 @@
 import { RPCServerOptions } from './types';
 import { DataOutput, DataInput } from '~/types';
+import { error } from '@karmic/core';
 
 export function createDefaults(): Required<Omit<RPCServerOptions, 'fallback'>> {
   return {
     children: true,
+    complete: {
+      name: 'EarlyCompleteError',
+      item: error({ label: 'ServerError' })
+    },
     parser: {
       serialize(data: object): DataOutput {
         return JSON.stringify(data);

packages/rpc/src/server/errors.ts

@@ -3,7 +3,9 @@ import {
   ErrorLabel,
   CollectionError,
   CollectionTree,
-  GeneralError
+  GeneralError,
+  ElementItem,
+  ErrorType
 } from '@karmic/core';
 import { RPCError } from '~/types';
 import { ErrorCodes } from '~/errors';
@@ -28,20 +30,31 @@ export const hash: { [P in ErrorLabel]: number } = {
   ServerTimeout: -32064
 };
 
+export type EnsureErrorType = Error | 'Server' | 'EarlyComplete';
+export type GetErrorType =
+  | 'ParseError'
+  | 'InvalidRequest'
+  | 'InternalError'
+  | PublicError;
+
 export function createEnsureError(
-  collection: CollectionTree
-): (error: Error) => PublicError {
+  collection: CollectionTree,
+  complete: ElementItem<ErrorType>
+): (error: EnsureErrorType) => PublicError {
   const id: GeneralError = 'ServerError';
   return function ensureError(error) {
     return error instanceof PublicError
       ? error
-      : new CollectionError(collection, id, error, true);
+      : new CollectionError(
+          collection,
+          error === 'EarlyComplete' ? complete.name : id,
+          null,
+          true
+        );
   };
 }
 
-export function getError(
-  err: 'ParseError' | 'InvalidRequest' | PublicError
-): RPCError {
+export function getError(err: GetErrorType): RPCError {
   if (typeof err === 'string') {
     const arr = Object.hasOwnProperty.call(ErrorCodes, err)
       ? ErrorCodes[err]

packages/rpc/src/server/types.ts

@@ -1,4 +1,8 @@
-import { QueryServiceImplementation } from '@karmic/core';
+import {
+  QueryServiceImplementation,
+  ElementItem,
+  ErrorType
+} from '@karmic/core';
 import { DataOutput, DataInput, DataParser } from '~/types';
 import { Observable } from 'rxjs';
 
@@ -12,6 +16,11 @@ export interface RPCServerOptions {
    * Serializer and deserializer
    */
   parser?: DataParser;
+  /**
+   * An error to emit if a stream completes before emitting any values.
+   * Defaults to an *EarlyCompleteError* `ServerError`.
+   */
+  complete?: ElementItem<ErrorType>;
   /**
    * A fallback service for adapters to use when the route is non existent.
    * Defaults to a `ClientNotFound` error throwing service.