import { Observable , Subject } from 'rx' ; import { Backend , FailedMutation , VersionedObject } from './backend' ; import { Version } from './tactical_store' ; /** A callback to handle resolving a mutation. */ export type AcceptHandler = ( version : string , data : Object ) => void ; /** A callback to handle failing a mutation. */ export type RejectHandler = ( rejection : Conflict | Failure ) => void ; /** A callback which indicates request processing is over (possibly with an error) */ export type Callback = ( err ? : any ) => void ; export type PublishHandler = ( key : Object , version : string , data : Object ) => void ; export class Conflict { version : string ; data : Object ; } export class Failure { constructor ( public reason , public info = { } ) { } } export interface MutationHandler { accept : AcceptHandler ; reject : RejectHandler ; publish : PublishHandler ; } /**
 * An interface that a backend must implement and provide to Tactical's implementation of
 * SocketServer.
 */ export interface BackendService { /**
     * Called when a request is received from the client. The `key` provided by the application
     * identifies the request. The backend is free to `publish` multiple responses for many different
     * keys.
     *
     *
     */ onRequest ( key , publish , callback ) : void ; /**
     * Called when a mutation for the given `key` is received from the client. This mutation is based
     * on
     * the version `base` and instructs the backend to set `key` to `value`.
     *
     * The backend is given a `handler` with a few different methods that can be called. If the
     * backend decides to accept the mutation, it should call `handler.accept` with the new version
     * and the mutated value. If the backend cannot accept the mutation, it should call
     * `handler.reject`,
     * the argument to which depends on the reason for rejection. Backends can reject for two reasons.
     * Either the mutation is against an outdated version, or the new object fails some validation
     * rule.
     *
     * If there is a newer version of the object in question, `handler.reject` should be called with a
     * `Conflict` object, which can be created from the version and value of the conflicting object.
     * Otherwise, it should be called with a `Failure` which takes an error message and an arbitrary
     * object that will be passed to the client for debugging purposes.
     *
     * `handler.publish` is available if the backend wishes to deliver updates to other keys that
     * happen
     * as a result of the mutation.
     *
     * `callback` should be called when the mutation has been handled, or earlier with any
     * error that might occur.
     */ onMutation ( key , base , value , handler , callback ) : void ; } /**
 * The type that is emitted on the 'data' stream from server to client.
 */ export type DataFrame = VersionedObject ; /**
 * The type that is emitted on the 'failure' stream from server to client.
 */ export type FailureFrame = FailedMutation ; /**
 * The type that is emitted on the 'request' stream from client to server.
 */ export interface RequestFrame { key : Object ; context ? : Object ; } /**
 * The type that is emitted on the 'mutation' stream from client to server.
 */ export interface MutationFrame { key : Object ; base : string ; data : Object ; context : Object ; } /**
 * A socket.io implementation of the SocketClient interface.
 */ export class SocketIOClient implements Backend { private _data : Subject < DataFrame > = new Subject < DataFrame > ( ) ; private _failures : Subject < FailureFrame > = new Subject < FailureFrame > ( ) ; /** Requires a socket.io socket to connect with. */ constructor ( public socket ) { this . _listen ( ) ; } data ( ) { return this . _data ; } failed ( ) { return this . _failures ; } mutate ( key , mutation , base , context ) { var frame = { key : key , base : base , data : mutation , context : context } ; this . socket . emit ( 'mutation' , frame ) ; } request ( key , cntxt ? ) { var frame = { key : key , context : ( cntxt ) ? cntxt : { } } ; this . socket . emit ( 'request' , frame ) ; } /**
     * Establishes listeners on the incoming client streams.
     */ private _listen ( ) { this . socket . on ( 'data' , ( frame ) => this . _data . onNext ( frame ) ) ; this . socket . on ( 'failure' , ( frame ) => this . _failures . onNext ( frame ) ) ; } } /**
 * A server that connects a `SocketIOClient` instance on the client to a `BackendService`
 */ export class SocketIOServer { constructor ( public service , public io ) { } /**
     * Send data to all connected clients.
     */ broadcastData ( key , base , data ) { var dataFrame = { key : key , version : base , data : data , mutationContext : null } ; this . io . emit ( 'data' , dataFrame ) ; } /**
     * Take over an incoming socket connection.
     */ public accept ( socket ) { var publish = ( key , version , data ) : void => { var dataFrame = { key : key , version : version , data : data , mutationContext : { } } ; socket . emit ( 'data' , dataFrame ) ; } ; socket . on ( 'request' , ( frame ) => { this . service . onRequest ( frame . key , publish , ( err ) => { if ( err ) throw err ; } ) ; } ) ; socket . on ( 'mutation' , ( frame ) => { var handler = { accept : ( version , data ) => { socket . emit ( 'data' , < DataFrame > { key : frame . key , version : version , data : data , mutationContext : frame . context } ) ; } , reject : ( rejection ) => { if ( rejection instanceof Conflict ) { socket . emit ( 'data' , < DataFrame > { key : frame . key , version : rejection . version , data : rejection . data } ) ; } else if ( rejection instanceof Failure ) { socket . emit ( 'failure' , < FailureFrame > { key : frame . key , reason : rejection . reason , debuggingInfo : rejection . info } ) ; } } , publish : publish } ; this . service . onMutation ( frame . key , frame . base , frame . data , handler , ( err ) => { if ( err ) throw err ; } ) ; } ) ; } }