__wip: wip

src/RelayPublic.js

@@ -16,6 +16,7 @@ const RelayContainer = require('RelayContainer');
 const RelayEnvironment = require('RelayEnvironment');
 const RelayInternals = require('RelayInternals');
 const RelayMutation = require('RelayMutation');
+const RelaySubscription = require('RelaySubscription');
 const RelayPropTypes = require('RelayPropTypes');
 const RelayQL = require('RelayQL');
 const RelayReadyStateRenderer = require('RelayReadyStateRenderer');
@@ -39,6 +40,7 @@ if (typeof global.__REACT_DEVTOOLS_GLOBAL_HOOK__ !== 'undefined') {
 const RelayPublic = {
   Environment: RelayEnvironment,
   Mutation: RelayMutation,
+  Subscription: RelaySubscription,
   PropTypes: RelayPropTypes,
   QL: RelayQL,
   ReadyStateRenderer: RelayReadyStateRenderer,

src/container/RelayContainer.js

@@ -160,6 +160,7 @@ function createContainerComponent(
           hasPartialData: this.hasPartialData.bind(this),
           pendingVariables: null,
           route,
+          subscribe: this.context.relay.subscribe,
           setVariables: this.setVariables.bind(this),
           variables: {},
         },

src/network-layer/default/RelayDefaultNetworkLayer.js

@@ -14,9 +14,12 @@
 
 import type RelayMutationRequest from 'RelayMutationRequest';
 import type RelayQueryRequest from 'RelayQueryRequest';
+import type RelaySubscriptionRequest from 'RelaySubscriptionRequest';
+import type {Subscription} from 'RelayTypes';
 
 const fetch = require('fetch');
 const fetchWithRetries = require('fetchWithRetries');
+const invariant = require('invariant')
 import type {InitWithRetries} from 'fetchWithRetries';
 
 type GraphQLError = {
@@ -89,6 +92,14 @@ class RelayDefaultNetworkLayer {
     )));
   }
 
+  sendSubscription(request: RelaySubscriptionRequest): Subscription {
+    invariant(
+      false,
+      'RelayDefaultNetworkLayer: `sendSubscription` is not implemented in the ' +
+      'default network layer.  A custom network layer must be injected.'
+    );
+  }
+
   supports(...options: Array<string>): boolean {
     // Does not support the only defined option, "defer".
     return false;

src/network/RelayNetworkLayer.js

@@ -13,10 +13,12 @@
 'use strict';
 
 import type RelayMutationRequest from 'RelayMutationRequest';
+import type RelaySubscriptionRequest from 'RelaySubscriptionRequest';
 const RelayProfiler = require('RelayProfiler');
 import type RelayQuery from 'RelayQuery';
 const RelayQueryRequest = require('RelayQueryRequest');
 import type {ChangeSubscription, NetworkLayer} from 'RelayTypes';
+import type {Subscription} from 'RelayTypes';
 
 const invariant = require('invariant');
 const resolveImmediate = require('resolveImmediate');
@@ -114,6 +116,29 @@ class RelayNetworkLayer {
     }
   }
 
+  sendSubscription(subscriptionRequest: RelaySubscriptionRequest): Subscription {
+    const implementation = this._getImplementation();
+
+    invariant(
+      typeof implementation.sendSubscription === 'function',
+      '%s: does not support subscriptions.  Expected `sendSubscription` to be ' +
+      'a function.',
+      implementation.constructor.name
+    );
+
+    const result = implementation.sendSubscription(subscriptionRequest);
+
+    invariant(
+      result && typeof result.dispose === 'function',
+      'RelayNetworkLayer: `sendSubscription` should return an object with a ' +
+      '`dispose` property that is a no-argument function.  This function is ' +
+      'called when the client unsubscribes from the subscription ' +
+      'and any network layer resources can be cleaned up.'
+    );
+
+    return result;
+  }
+
   supports(...options: Array<string>): boolean {
     const implementation = this._getImplementation();
     return implementation.supports(...options);

src/network/RelaySubscriptionRequest.js

@@ -0,0 +1,115 @@
+/**
+ * Copyright (c) 2013-present, Facebook, Inc.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree. An additional grant
+ * of patent rights can be found in the PATENTS file in the same directory.
+ *
+ * @providesModule RelaySubscriptionRequest
+ * @flow
+ */
+
+'use strict';
+
+import type {PrintedQuery} from 'RelayInternalTypes';
+import type RelayQuery from 'RelayQuery';
+import type {
+  SubscriptionCallbacks,
+  SubscriptionResult,
+  Variables
+} from 'RelayTypes';
+
+const printRelayQuery = require('printRelayQuery');
+
+/**
+ * @internal
+ *
+ * Instances of these are made available via `RelayNetworkLayer.sendSubscription`.
+ */
+class RelaySubscriptionRequest {
+  _subscription: RelayQuery.Subscription;
+  _printedQuery: ?PrintedQuery;
+
+  constructor(
+    subscription: RelayQuery.Subscription,
+    callbacks: SubscriptionCallbacks,
+  ) {
+    this._subscription = subscription;
+    this._callbacks = callbacks;
+  }
+
+  /**
+   * @public
+   *
+   * Gets a string name used to refer to this request for printing debug output.
+   */
+  getDebugName(): string {
+    return this._subscription.getName();
+  }
+
+  /**
+   * @public
+   *
+   * Gets the variables used by the subscription. These variables should be
+   * serialized and sent in the GraphQL request.
+   */
+  getVariables(): Variables {
+    return this._getPrintedQuery().variables;
+  }
+
+  /**
+   * @public
+   *
+   * Gets a string representation of the GraphQL subscription.
+   */
+  getQueryString(): string {
+    return this._getPrintedQuery().text;
+  }
+
+  /**
+   * @public
+   * @unstable
+   */
+  getSubscription(): RelayQuery.Subscription {
+    return this._subscription;
+  }
+
+  /**
+   * @public
+   * @unstable
+   */
+  onCompleted(): void {
+    return this._callbacks && this._callbacks.onComplete();
+  }
+
+  /**
+   * @public
+   * @unstable
+   */
+  onNext(payload: SubscriptionResult): void {
+    return this._callbacks && this._callbacks.onNext(payload);
+  }
+
+  /**
+   * @public
+   * @unstable
+   */
+  onError(error: Error): void {
+    return this._callbacks && this._callbacks.onError(error);
+  }
+
+  /**
+   * @private
+   *
+   * Returns the memoized printed query.
+   */
+  _getPrintedQuery(): PrintedQuery {
+    if (!this._printedQuery) {
+      this._printedQuery = printRelayQuery(this._subscription);
+    }
+    return this._printedQuery;
+  }
+}
+
+module.exports = RelaySubscriptionRequest;

src/store/RelayEnvironment.js

@@ -18,8 +18,10 @@ import type RelayMutationTransaction from 'RelayMutationTransaction';
 import type {MutationCallback, QueryCallback} from 'RelayNetworkLayer';
 import type RelayQuery from 'RelayQuery';
 import type RelayQueryTracker from 'RelayQueryTracker';
+import type RelaySubscription from 'RelaySubscription';
 const RelayQueryResultObservable = require('RelayQueryResultObservable');
 const RelayStoreData = require('RelayStoreData');
+import type {Subscription} from 'RelayTypes';
 import type {TaskScheduler} from 'RelayTaskQueue';
 import type {ChangeSubscription, NetworkLayer} from 'RelayTypes';
 
@@ -28,13 +30,16 @@ const readRelayQueryData = require('readRelayQueryData');
 const relayUnstableBatchedUpdates = require('relayUnstableBatchedUpdates');
 const warning = require('warning');
 
+const noop = function () {}
+
 import type {
   Abortable,
   Observable,
   RelayMutationTransactionCommitCallbacks,
   ReadyStateChangeCallback,
   StoreReaderData,
   StoreReaderOptions,
+  SubscriptionCallbacks,
 } from 'RelayTypes';
 
 import type {
@@ -94,6 +99,10 @@ class RelayEnvironment {
     mutation: RelayMutation<any>,
     callbacks?: RelayMutationTransactionCommitCallbacks
   ) => RelayMutationTransaction;
+  subscribe: (
+    subscription: RelaySubscription<any>,
+    callbacks?: SubscriptionCallbacks
+  ) => Subscription;
   _storeData: RelayStoreData;
 
   constructor(storeData?: RelayStoreData) {
@@ -103,6 +112,7 @@ class RelayEnvironment {
     );
     this.applyUpdate = this.applyUpdate.bind(this);
     this.commitUpdate = this.commitUpdate.bind(this);
+    this.subscribe = this.subscribe.bind(this);
   }
 
   /**
@@ -269,6 +279,24 @@ class RelayEnvironment {
       .commit();
   }
 
+  /**
+   * Adds a subscription to the store and subscribes to it immediately.
+   * Returns the Rx Subscription (disposable).
+   */
+  subscribe(
+    subscription: RelaySubscription<any>,
+    callbacks?: SubscriptionCallbacks
+  ): Subscription {
+    callbacks = callbacks || {
+      onCompleted: noop,
+      onError: noop,
+      onNext: noop,
+    };
+    subscription.bindEnvironment(this);
+    const subscriptionObservable = this._storeData.getSubscriptionObserver().observe(subscription)
+    return subscriptionObservable.subscribe(callbacks);
+  }
+
   /**
    * @deprecated
    *
@@ -287,4 +315,4 @@ class RelayEnvironment {
   }
 }
 
-module.exports = RelayEnvironment;
+module.exports = RelayEnvironment;

src/store/RelayStoreData.js

@@ -51,6 +51,7 @@ const forEachObject = require('forEachObject');
 const mapObject = require('mapObject');
 const invariant = require('invariant');
 const generateForceIndex = require('generateForceIndex');
+const RelaySubscriptionObserver = require('RelaySubscriptionObserver');
 const {
   restoreFragmentDataFromCache,
   restoreQueriesDataFromCache,
@@ -95,6 +96,7 @@ class RelayStoreData {
   _queryTracker: ?RelayQueryTracker;
   _queryRunner: GraphQLQueryRunner;
   _rangeData: GraphQLStoreRangeUtils;
+  _relaySubscriptionObserver: RelaySubscriptionObserver;
   _rootCallMap: RootCallMap;
   _taskQueue: RelayTaskQueue;
 
@@ -128,13 +130,14 @@ class RelayStoreData {
       {cachedRootCallMap, rootCallMap},
       nodeRangeMap
     );
+    this._rangeData = rangeData;
     this._records = records;
     this._recordStore = new RelayRecordStore(
       {records},
       {rootCallMap},
       nodeRangeMap
     );
-    this._rangeData = rangeData;
+    this._relaySubscriptionObserver = new RelaySubscriptionObserver(this);
     this._rootCallMap = rootCallMap;
     this._taskQueue = new RelayTaskQueue();
   }
@@ -535,6 +538,10 @@ class RelayStoreData {
     return this._taskQueue;
   }
 
+  getSubscriptionObserver(): RelaySubscriptionObserver {
+    return this._relaySubscriptionObserver;
+  }
+
   /**
    * @deprecated
    *

src/store/RelaySubscriptionObserver.js

@@ -0,0 +1,65 @@
+/**
+ * Copyright (c) 2013-present, Facebook, Inc.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree. An additional grant
+ * of patent rights can be found in the PATENTS file in the same directory.
+ *
+ * @providesModule RelaySubscriptionObserver
+ * @flow
+ */
+
+'use strict';
+
+import type RelayStoreData from 'RelayStoreData';
+import type RelaySubscription from 'RelaySubscription';
+const RelaySubscriptionObservable = require('RelaySubscriptionObservable');
+
+/**
+ * Central class for observing subscriptions. This class keeps a central reference to all subscriptions
+ */
+class RelaySubscriptionObserver {
+  _storeData: RelayStoreData;
+  _observables: Array<RelaySubscriptionObservable>;
+
+  constructor(
+    storeData: RelayStoreData,
+  ) {
+    this._observables = []
+    this._storeData = storeData;
+  }
+
+  /**
+   * Observe a subscription, ensures that subscriptions are
+   */
+  observe(
+    subscription: RelaySubscription
+  ): RelaySubscriptionObservable {
+    let observable = this._observables.find(observable => observable.getSubscription() === subscription);
+    if (!observable) {
+      observable = new RelaySubscriptionObservable(
+        this,
+        this._storeData,
+        subscription,
+      )
+      this._observables.push(observable);
+    }
+    return observable;
+  }
+
+  /**
+   * Unobserve a subscription from it's observable
+   */
+  unobserve(
+    subscription: RelaySubscription
+  ): void {
+    const observableIndex = this._observables.findIndex(observable => observable.getSubscription() === subscription);
+    if (observableIndex >= 0) {
+      this._observables[observableIndex].unobserve();
+      this._observables.splice(observableIndex, 1);
+    }
+  }
+}
+
+module.exports = RelaySubscriptionObserver;