Apollo 服务器订阅中间件拦截请求主体并向不同的服务器请求相同的请求

Question

我有一个带道具的 NodeJS Apollo 服务器

// Graphql server configuration.
  const graphqlServerConfig: ApolloServerExpressConfig = {
    schema: TEST_MODE ? mergedSchema : mergedSchemaWithPermissions,
    context: ({ req, res, connection }): GraphQLContext => {
      return {
        req,
        res,
        csctx,
      };
    },
    subscriptions: {
      path: '/subscriptions',
      keepAlive: 1000,
    },
    playground: {
      endpoint: '/core/graphql/',
      subscriptionEndpoint: '/core/graphql/subscriptions',
      settings: {
        'request.credentials': 'include',
      },
    },
  };

我需要一个中间件订阅，它可以使用我收到的相同请求订阅远程服务器。基本上，简而言之，UI 将请求订阅。在

connectionParams

的基础上，我需要在服务器端创建一个新的websocket客户端，并返回我们从那里得到的任何响应。

Answer 1

您应该考虑模式拼接，而不是拦截请求并解析它以找出将它委托到哪里。

此示例从远程服务器获取 graphql 模式，并使其在您的服务器上可用，同时将其操作（查询、变更和订阅）的解析委托给远程服务器。

确保远程服务器实现与您的服务器通信的相同协议，在我的例子中是 GraphQL over WebSocket Protocol 因为我的远程服务器是带有 absinthe_graphql_ws.

的 Phoenix 服务器

import { Client, createClient } from "graphql-ws";
import { createServer } from "http";
import { WebSocket, WebSocketServer } from "ws";
import { useServer } from "graphql-ws/lib/use/ws";
import { GraphQLSchema, print } from "graphql";
import { AsyncExecutor, observableToAsyncIterable } from "@graphql-tools/utils";
import { schemaFromExecutor, wrapSchema } from "@graphql-tools/wrap";

import { ApolloServer } from "@apollo/server";
import { expressMiddleware } from "@apollo/server/express4";
import { ApolloServerPluginDrainHttpServer } from "@apollo/server/plugin/drainHttpServer";

import express from "express";
import bodyParser from "body-parser";
import cors from "cors";

const HTTP_GRAPHQL_ENDPOINT = "http://localhost:4001/api";
const WS_GRAPHQL_ENDPOINT = "ws://localhost:4001/api/graphql-ws";

const app = express();
const httpServer = createServer(app);

const subscriptionClient = createClient({
  url: WS_GRAPHQL_ENDPOINT,
  webSocketImpl: WebSocket,
});

const schema = await getRemoteSchema(subscriptionClient);

const wsServer = new WebSocketServer({
  server: httpServer,
  path: "/graphql",
});

const serverCleanup = useServer({ schema }, wsServer);

const server = new ApolloServer({
  schema,
  plugins: [
    ApolloServerPluginDrainHttpServer({ httpServer }),
    {
      async serverWillStart() {
        return {
          async drainServer() {
            await serverCleanup.dispose();
          },
        };
      },
    },
  ],
});

await server.start();

app.use(
  "/graphql",
  cors<cors.CorsRequest>(),
  bodyParser.json(),
  expressMiddleware(server)
);

httpServer.listen(1338, () => {
  console.info('Server is running on http://localhost:1338/graphql')
});

async function getRemoteSchema(
  subscriptionClient: Client
): Promise<GraphQLSchema> {
  const httpExecutor: AsyncExecutor = async ({
    document,
    variables,
    operationName,
    extensions,
  }) => {
    const query = print(document);
    const fetchResult = await fetch(HTTP_GRAPHQL_ENDPOINT, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Accept: "application/json",
      },
      body: JSON.stringify({ query, variables, operationName, extensions }),
    });
    return fetchResult.json();
  };

  const wsExecutor: AsyncExecutor = async ({
    document,
    variables,
    operationName,
    extensions,
  }) =>
    observableToAsyncIterable({
      subscribe: (observer) => ({
        unsubscribe: subscriptionClient.subscribe(
          {
            query: print(document),
            variables: variables as Record<string, any>,
            operationName,
            extensions,
          },
          {
            next: (data) => observer.next?.(data as unknown),
            error(err) {
              if (!observer.error) return;
              if (err instanceof Error) {
                observer.error(err);
              } else if (err instanceof CloseEvent) {
                observer.error(
                  new Error(`Socket closed with event ${err.code}`)
                );
              } else if (Array.isArray(err)) {
                // GraphQLError[]
                observer.error(
                  new Error(err.map(({ message }) => message).join(", "))
                );
              }
            },
            complete: () => observer.complete?.(),
          }
        ),
      }),
    });

  const executor: AsyncExecutor = async (executorRequest) => {
    if (executorRequest.operationType === "subscription") {
      return wsExecutor(executorRequest);
    }

    return httpExecutor(executorRequest);
  };

  const schema = wrapSchema({
    schema: await schemaFromExecutor(executor),
    executor,
  });

  return schema;
}

---

使用 graphql-yoga 代替 ApolloServer 使代码更简洁：

// ... all imports except apollo & related, express
import { createYoga } from 'graphql-yoga'

// ...
const schema = await getRemoteSchema(subscriptionClient);
const yoga = createYoga({ schema: schema })
const server = createServer(yoga)

server.listen(1338, () => {
   console.info('Server is running on http://localhost:1338/graphql')
})

希望对您有所帮助！

---

确切的依赖版本：

  "dependencies": {
    "@apollo/server": "^4.7.1",
    "@graphql-tools/schema": "^9.0.19",
    "@graphql-tools/wrap": "^9.4.2",
    "body-parser": "^1.20.2",
    "cors": "^2.8.5",
    "cross-fetch": "^3.1.5",
    "graphql-ws": "^5.12.1",
    "graphql-yoga": "^3.9.1",
    "ws": "^8.13.0"
  },

来源：

• 远程子模式和执行者
• Apollo Server 中的订阅