中继现代:将websocket连接到网络层

Question

对于如何将中继现代网络层与websocket实例连接起来，我遇到了一些问题。

我目前正在实例化一个websocket实例，如下所示：

const subscriptionWebSocket = new ReconnectingWebSocket('ws://url.url/ws/subscriptions/', null, options);

我正在指定订阅并创建一个新的requestSubscription实例

const subscription = graphql`
  subscription mainSubscription {
    testData {
      anotherNode {
        data
      }
    }
  }
`;

requestSubscription(
  environment,
  {
    subscription,
    variables: {},
    onComplete: () => {...},
    onError: (error) => {...},
    onNext: (response) => {...},
    updater: (updaterStoreConfig) => {...},
  },
);

然后允许我发送任何订阅请求：

function subscriptionHandler(subscriptionConfig, variables, cacheConfig, observer) {
  subscriptionWebSocket.send(JSON.stringify(subscriptionConfig.text));

  return {
    dispose: () => {
      console.log('subscriptionHandler: Disposing subscription');
    },
  };
}

const network = Network.create(fetchQuery, subscriptionHandler);

通过我的服务器(目前使用Graphene-python)，我可以在服务器上解释接收到的消息。

但是，我遇到的问题是如何响应订阅；例如，当我的DB中的某些内容发生变化时，我希望生成一个响应并返回给任何潜在的订阅者。

问题是，如何将onMessage事件从websocket实例连接到中继现代网络层？我已经浏览过中继的源代码，但似乎找不出什么回调，或者应该实现onreceive的方法。

任何小贴士都会很感激。

Answer 1

在这个线程中找到帮助之后，我将写下我是如何处理这个问题的。它可能对其他人有用。这非常依赖于您选择的服务器端解决方案。

我的方法：

首先，我构建了一个SubscriptionHandler，它将通过SubscriptionHandler#setupSubscription来处理requestStream#subscribeFunction。

(SubscriptionHandler#receiveSubscriptionPayload)实例化一个WebSocket (使用ReconnectingWebSockets的自定义版本)，并将onmessage事件附加到内部方法WebSocket，该方法将向相应的请求添加有效负载。

我们通过SubscriptionHandler#newSubscription创建新的订阅，它将使用内部属性SubscriptionHandler.subscriptions添加该订阅的键控条目(对查询和变量使用MD5-散列util )；这意味着对象将显示为：

SubscriptionHandler.subscriptions = {
  [md5hash]: {
    query: QueryObject,
    variables: SubscriptionVariables,
    observer: Observer (contains OnNext method)
}

每当服务器发送订阅响应时，将调用SubscriptionHandler#receiveSubscriptionPayload方法，它将通过使用查询/变量md5哈希来识别有效负载属于什么订阅，然后使用SubscriptionHandler.subscriptions观察者onNext方法。

此方法要求服务器返回如下消息：

export type ServerResponseMessageParsed = {
  payload: QueryPayload,
  request: {
    query: string,
    variables: Object,
  }
}

我不知道这是否是处理订阅的一种很好的方式，但现在它适用于我当前的设置。

SubscriptionHandler.js

class SubscriptionHandler {
  subscriptions: Object;
  subscriptionEnvironment: RelayModernEnvironment;
  websocket: Object;

  /**
   * The SubscriptionHandler constructor. Will setup a new websocket and bind
   * it to internal method to handle receving messages from the ws server.
   *
   * @param  {string} websocketUrl      - The WebSocket URL to listen to.
   * @param  {Object} webSocketSettings - The options object.
   *                                      See ReconnectingWebSocket.
   */
  constructor(websocketUrl: string, webSocketSettings: WebSocketSettings) {
    // All subscription hashes and objects will be stored in the
    // this.subscriptions attribute on the subscription handler.
    this.subscriptions = {};

    // Store the given environment internally to be reused when registering new
    // subscriptions. This is required as per the requestRelaySubscription spec
    // (method requestSubscription).
    this.subscriptionEnvironment = null;

    // Create a new WebSocket instance to be able to receive messages on the
    // given URL. Always opt for default protocol for the RWS, second arg.
    this.websocket = new ReconnectingWebSocket(
      websocketUrl,
      null,  // Protocol.
      webSocketSettings,
    );

    // Bind an internal method to handle incoming messages from the websocket.
    this.websocket.onmessage = this.receiveSubscriptionPayload;
  }

  /**
   * Method to attach the Relay Environment to the subscription handler.
   * This is required as the Network needs to be instantiated with the
   * SubscriptionHandler's methods, and the Environment needs the Network Layer.
   *
   * @param  {Object} environment - The apps environment.
   */
  attachEnvironment = (environment: RelayModernEnvironment) => {
    this.subscriptionEnvironment = environment;
  }

  /**
   * Generates a hash from a given query and variable pair. The method
   * used is a recreatable MD5 hash, which is used as a 'key' for the given
   * subscription. Using the MD5 hash we can identify what subscription is valid
   * based on the query/variable given from the server.
   *
   * @param  {string} query     - A string representation of the subscription.
   * @param  {Object} variables - An object containing all variables used
   *                              in the query.
   * @return {string}             The MD5 hash of the query and variables.
   */
  getHash = (query: string, variables: HashVariables) => {
    const queryString = query.replace(/\s+/gm, '');
    const variablesString = JSON.stringify(variables);
    const hash = md5(queryString + variablesString).toString();
    return hash;
  }

  /**
   * Method to be bound to the class websocket instance. The method will be
   * called each time the WebSocket receives a message on the subscribed URL
   * (see this.websocket options).
   *
   * @param  {string} message - The message received from the websocket.
   */
  receiveSubscriptionPayload = (message: ServerResponseMessage) => {
    const response: ServerResponseMessageParsed = JSON.parse(message.data);
    const { query, variables } = response.request;
    const hash = this.getHash(query, variables);

    // Fetch the subscription instance from the subscription handlers stored
    // subscriptions.
    const subscription = this.subscriptions[hash];

    if (subscription) {
      // Execute the onNext method with the received payload after validating
      // that the received hash is currently stored. If a diff occurs, meaning
      // no hash is stored for the received response, ignore the execution.
      subscription.observer.onNext(response.payload);
    } else {
      console.warn(Received payload for unregistered hash: ${hash});
    }
  }

  /**
   * Method to generate new subscriptions that will be bound to the
   * SubscriptionHandler's environment and will be stored internally in the
   * instantiated handler object.
   *
   * @param {string} subscriptionQuery - The query to subscribe to. Needs to
   *                                     be a validated subscription type.
   * @param {Object} variables         - The variables for the passed query.
   * @param {Object} configs           - A subscription configuration. If
   *                                     override is required.
   */
  newSubscription = (
      subscriptionQuery: GraphQLTaggedNode,
      variables: Variables,
      configs: GraphQLSubscriptionConfig,
  ) => {
    const config = configs || DEFAULT_CONFIG;

    requestSubscription(
      this.subscriptionEnvironment,
      {
        subscription: subscriptionQuery,
        variables: {},
        ...config,
      },
    );
  }

  setupSubscription = (
    config: ConcreteBatch,
    variables: Variables,
    cacheConfig: ?CacheConfig,
    observer: Observer,
  ) => {
    const query = config.text;

    // Get the hash from the given subscriptionQuery and variables. Used to
    // identify this specific subscription.
    const hash = this.getHash(query, variables);

    // Store the newly created subscription request internally to be re-used
    // upon message receival or local data updates.
    this.subscriptions[hash] = { query, variables };

    const subscription = this.subscriptions[hash];
    subscription.observer = observer;

    // Temp fix to avoid WS Connection state.
    setTimeout(() => {
      this.websocket.send(JSON.stringify({ query, variables }));
    }, 100);
  }
}

const subscriptionHandler = new SubscriptionHandler(WS_URL, WS_OPTIONS);

export default subscriptionHandler;

Answer 2

我已经成功地订阅了中继现代工作，并希望分享我的最低设置，也许这是有益的人！

请注意，我不是使用WebSocket，而是使用subscriptions-transport-ws中可以找到的SubscriptionClient来管理到服务器的连接。

下面是我的最低设置代码：

Environment.js

import { SubscriptionClient } from 'subscriptions-transport-ws'
const {
  Environment,
  Network,
  RecordSource,
  Store,
} = require('relay-runtime')
const store = new Store(new RecordSource())


const fetchQuery = (operation, variables) => {
  return fetch('https://api.graph.cool/relay/v1/__GRAPHCOOL_PROJECT_ID__', {
    method: 'POST',
    headers: {
      'Accept': 'application/json',
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      query: operation.text,
      variables,
    }),
  }).then(response => {
    return response.json()
  })
}

const websocketURL = 'wss://subscriptions.graph.cool/v1/__GRAPHCOOL_PROJECT_ID__'

function setupSubscription(
  config,
  variables,
  cacheConfig,
  observer,
) {
  const query = config.text

  const subscriptionClient = new SubscriptionClient(websocketURL, {reconnect: true})
  const id = subscriptionClient.subscribe({query, variables}, (error, result) => {
    observer.onNext({data: result})
  })
}

const network = Network.create(fetchQuery, setupSubscription)
const environment = new Environment({
  network,
  store,
})

export default environment

NewLinkSubscription.js

import {
  graphql,
  requestSubscription
} from 'react-relay'
import environment from '../Environment'

const newLinkSubscription = graphql`
  subscription NewLinkSubscription {
    Link {
      mutation
      node {
        id
        description
        url
        createdAt
        postedBy {
          id
          name
        }
      }
    }
  }
`

export default (onNext, onError, onCompleted, updater) => {

  const subscriptionConfig = {
    subscription: newLinkSubscription,
    variables: {},
    onError,
    onNext,
    onCompleted,
    updater
  }

  requestSubscription(
    environment,
    subscriptionConfig
  )

}

现在，您可以简单地使用导出的函数进行订阅。例如，在componentDidMount中的一个React组件中，我现在可以执行以下操作：

componentDidMount() {
  NewLinkSubscription(
    response => console.log(`Received data: `, response),
    error => console.log(`An error occurred:`, error),
    () => console.log(`Completed`)
  )
}

请注意，SubscriptionClient只能在服务器实现这协议的情况下使用！

如果您想了解更多信息，请查看完整的堆栈如何使用GraphQL教程，它详细描述了如何使订阅与中继现代一起工作。

Answer 3

对于最近偶然发现这一点的人来说，我在上述两种解决方案中都没有成功，因为最近在所涉及的库中进行了更新。然而，它们是一个很好的起点，我举了一个基于官方接力现代todo例子的小例子，它非常简约，使用来自阿波罗的帮助图书馆，但与中继现代很好地工作：

https://github.com/jeremy-colin/relay-examples-subscription

它包括服务器和客户端。

希望能帮上忙