GraphQL订阅使用Express-GraphQL和graphql-订阅为空

Question

我正在使用TypeScript，并拥有服务器和客户端应用程序。下面是服务器代码。

服务器代码

import express, { Express } from "express";
import { graphqlHTTP } from "express-graphql";
import { buildSchema } from "type-graphql";
import { TaskResolver } from "./resolvers/task.resolver";
import { pgDatasource } from "./configs/db.config";
import { SeatBandingResolver } from "./resolvers/seatBanding.resolver";
import { GuestChatResolver } from "./resolvers/guestChat.resolver";
import { RateResolver } from "./resolvers/rate.resolver";
import { YearResolver } from "./resolvers/year.resolver";
import { ImplementationRateResolver } from "./resolvers/implementationRate.resolver";
import { UserResolver } from "./resolvers/user.resolver";
import { ReportResolver } from "./resolvers/report.resolver";

// Subscriptions
const ws = require("ws");
const { useServer } = require("graphql-ws/lib/use/ws");
const { execute, subscribe } = require("graphql");

const main = async () => {
  const app: Express = express();
  try {
    //connect to db
    await pgDatasource.initialize();
  } catch (err) {
    throw err;
  }

  //build gql schema
  let schema = await buildSchema({
    resolvers: [
      SeatBandingResolver,
      GuestChatResolver,
      RateResolver,
      YearResolver,
      ImplementationRateResolver,
      UserResolver,
    ],
    validate: false,
    // pubSub: new PubSub()
  });
  let schemaDoc = await buildSchema({
    resolvers: [ReportResolver],
    validate: false,
  });

  //ql schema for report
  const docServer = graphqlHTTP((req, res) => {
    return {
      schema: schemaDoc,
      graphiql: true,
      context: {
        req: req,
        header: req.headers,
      },
    };
  });
  //setting a graphql server instance
  const graphqServer = graphqlHTTP((req, res, graphQLParams) => {
    return {
      schema,
      context: {
        req: req,
        header: req.headers,
      },
      graphiql: true,
    };
  });

  app.use(cors());

  //graphql endpoint : change it to backend
  app.use("/graphql", graphqServer);

  //for report : change name to google api
  app.use("/doc", docServer);

  //test route
  app.get("/", (req, res) => {
    res.json({
      message: "Hello world",
    });
  });

  let server = app.listen(3001, () => {
    console.log("server started");

    const wsServer = new ws.WebSocketServer({
      host: "localhost",
      // server,
      path: "/graphql",
      port: 3001,
    });

    useServer(
      {
        schema,
        execute,
        subscribe,
        onConnect: (ctx) => {
          console.log("Connect");
        },
        onSubscribe: (ctx, msg) => {
          console.log("Subscribe");
        },
        onNext: (ctx, msg, args, result) => {
          console.debug("Next");
        },
        onError: (ctx, msg, errors) => {
          console.error("Error");
        },
        onComplete: (ctx, msg) => {
          console.log("Complete");
        },
      },
      wsServer
    );
  });
};

//starting a server
main()
  .then(async (_) => {
    // await addColumn()
  })
  .catch((err) => {
    console.log(err);
  });

客户端的订阅代码

import { Year } from "../entities/year.entity";
import { NewYear } from "../inputs/addYear.input";
import {
  Arg,
  Ctx,
  Field,
  Int,
  Mutation,
  ObjectType,
  Query,
  Resolver,
  Root,
  Subscription,
  UseMiddleware,
} from "type-graphql";
import { Request } from "express";
import { Response } from "../helpers/response.helper";
import { Pagination } from "../inputs/pagination.input";
import { isAuth } from "../helpers/auth.helper";
import { PubSub, PubSubEngine } from "graphql-subscriptions";
const pubSub = new PubSub();

@ObjectType()
class MessagePayload {
  @Field()
  message: string;
}
@Resolver(() => Year)
export class YearResolver {
  @Mutation(() => String)
  async sendMessage(@Arg("message") message: string): Promise<string> {
    console.log("in send subscription");
    pubSub.publish("MESSAGE_NOTIFICATION", { message });
    return message;
  }

  //calling the subscription
  @Subscription(() => MessagePayload || null, {
    topics: "MESSAGE_NOTIFICATION",
  })
  async receiveMessage(
    @Root() root: MessagePayload
  ): Promise<MessagePayload | null> {
    console.log("in publisher");
    console.log(root, "in recieve message");
    pubSub.asyncIterator("MESSAGE_NOTIFICATION");
    return { message: "hello from the subscription" };
  }
}

我在这里面临的问题是订阅不能正常工作，而且数据总是为空。

有人能帮我辨认出我在这里错过了什么吗？

谢谢。

Answer 1

我不确定100%，因为您的代码描述有点混乱，但是看起来您应该在方法receiveMessage中返回pubSub.asyncIterator('MESSAGE_NOTIFICATION')。调用此方法是为了在选定的通道(MESSAGE_NOTIFICATION)开始向客户端发送流消息，而不是发送消息。若要发送消息，请使用pubsub。当然你也应该改变打字方式。

您可以找到一个类似的实现这里。