GraphQL实践10——Netflix Dgs Graphql异步订阅

		  <dependency>
            <groupId>com.netflix.graphql.dgs</groupId>
            <artifactId>graphql-dgs-subscriptions-websockets-autoconfigure</artifactId>
        </dependency>

@Slf4j
@DgsComponent
@RequiredArgsConstructor
public class ActorDataFetcher {

	private final ActorAssembler actorAssembler;

	private FluxSink<Actor> actorStream;

	private ConnectableFlux<Actor> actorPublisher;

	@PostConstruct
	private void createActorPublisher() {
		Flux<Actor> publisher = Flux.create(emitter -> {
			actorStream = emitter;
		});

		actorPublisher = publisher.publish();
		actorPublisher.connect();
	}

	@DgsMutation
	public Actor addActor(@InputArgument SubmitActor actor) {
		Actor actorEntity = actorAssembler.convert(actor);
		actorEntity.setActorId(10);
		actorEntity.setLastUpdate(DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
		actorStream.next(actorEntity);
		log.info("服务端创建演员：{}", actor);
		return actorEntity;
	}

	/**
	 * 异步通知
	 *
	 * @return
	 */
	@DgsSubscription
	public Publisher<Actor> actorAdded() {
		return actorPublisher;
	}
}

@Slf4j
@SpringBootTest
class ActorDataFetcherTest {

	@Autowired
	private DgsQueryExecutor dgsQueryExecutor;

	@Test
	void addActor() {

		GraphQLQueryRequest graphQLQueryRequest = new GraphQLQueryRequest(
				AddActorGraphQLQuery.newRequest()
						.actor(
								SubmitActor.newBuilder()
										.firstName("fu")
										.lastName("jiayang")
										.build())
						.build(),
				new AddActorProjectionRoot()
						.actorId()
						.firstName()
						.lastUpdate());
		dgsQueryExecutor.execute(graphQLQueryRequest.serialize());
	}

	@Test
	void actorAdded() {
		ExecutionResult executionResult = dgsQueryExecutor.execute("""
				subscription {
				  actorAdded {
				    actorId
				    firstName
				    lastName
				    lastUpdate
				  }
				}
				""");
		Publisher<ExecutionResult> reviewPublisher = executionResult.getData();
		List<Actor> actors = new CopyOnWriteArrayList<>();

		reviewPublisher.subscribe(new Subscriber<>() {
			@Override
			public void onSubscribe(Subscription s) {
				s.request(2);
			}

			@Override
			public void onNext(ExecutionResult executionResult) {
				if (executionResult.getErrors().size() > 0) {
					executionResult.getErrors().forEach(error -> log.error(error.toString()));
				}
				Map<String, Object> actorResult = executionResult.getData();
				Actor actor = new ObjectMapper().convertValue(actorResult.get("actorAdded"), Actor.class);
             log.info("客户端监听到演员新增：{}",actor);
				actors.add(actor);
			}

			@Override
			public void onError(Throwable t) {
				t.printStackTrace();
			}

			@Override
			public void onComplete() {
			}
		});

		addActor();
		addActor();

		assertThat(actors.size()).isEqualTo(2);
	}
}