@Nullable
    public void handleText(String event, WebSocketWriter out) {
        if (!out.isOpen()) {
            return;
        }

        try {
            final Map<String, Object> eventMap = parseJsonString(event, JSON_MAP);
            final String type = toStringFromJson(eventMap.get("type"));
            if (type == null) {
                throw new GraphqlWebSocketCloseException(4400, "type is required");
            }
            final String id;

            switch (type) {
                case "connection_init":
                    if (connectionInitiated) {
                        // Already initiated, that's an error
                        throw new GraphqlWebSocketCloseException(4429, "Already initiated");
                    }
                    final Object rawPayload = eventMap.get("payload");
                    if (rawPayload != null) {
                        connectionCtx = toMapFromJson(rawPayload);
                    }
                    connectionInitiated = true;
                    writeConnectionAck(out);
                    break;
                case "ping":
                    writePong(out);
                    break;
                case "pong":
                    break;
                case "subscribe":
                    ensureInitiated();
                    id = toStringFromJson(eventMap.get("id"));
                    if (id == null) {
                        throw new GraphqlWebSocketCloseException(4400, "id is required");
                    }
                    final Map<String, Object> payload = toMapFromJson(eventMap.get("payload"));
                    try {
                        if (graphqlSubscriptions.containsKey(id)) {
                            // Subscription already exists
                            throw new GraphqlWebSocketCloseException(4409, "Already subscribed");
                        }
                        final String operationName = toStringFromJson(payload.get("operationName"));
                        final String query = toStringFromJson(payload.get("query"));
                        final Map<String, Object> variables = toMapFromJson(payload.get("variables"));
                        final Map<String, Object> extensions = toMapFromJson(payload.get("extensions"));

                        final ExecutionInput executionInput =
                                ExecutionInput.newExecutionInput()
                                              .graphQLContext(connectionCtx)
                                              .graphQLContext(upgradeCtx)
                                              .query(query)
                                              .variables(variables)
                                              .operationName(operationName)
                                              .extensions(extensions)
                                              .dataLoaderRegistry(dataLoaderRegistryFunction.apply(ctx))
                                              .build();

                        final CompletableFuture<ExecutionResult> future =
                                graphqlExecutor.executeGraphql(ctx, executionInput);

                        future.handleAsync((executionResult, throwable) -> {
                            handleExecutionResult(out, id, executionResult, throwable);
                            return null;
                        }, ctx.eventLoop());
                    } catch (GraphqlWebSocketCloseException e) {
                        logger.debug("Error handling subscription", e);
                        // Also cancel subscription if present before closing websocket
                        final ExecutionResultSubscriber s = graphqlSubscriptions.remove(id);
                        if (s != null) {
                            s.setCompleted();
                        }
                        out.close(e.getWebSocketCloseStatus());
                    } catch (Exception e) {
                        logger.debug("Error handling subscription", e);
                        // Unknown but possibly recoverable error
                        writeError(out, id, e);
                        return;
                    }
                    break;
                case "complete":
                    ensureInitiated();
                    // Read id and remove that subscription
                    id = toStringFromJson(eventMap.get("id"));
                    if (id == null) {
                        throw new GraphqlWebSocketCloseException(4400, "id is required");
                    }
                    final ExecutionResultSubscriber s = graphqlSubscriptions.remove(id);
                    if (s != null) {
                        s.setCompleted();
                    }
                    return;
                default:
                    final String reasonPhrase = maybeTruncate("Unknown event type: " + type);
                    assert reasonPhrase != null;
                    throw new GraphqlWebSocketCloseException(4400, reasonPhrase);
            }
        } catch (GraphqlWebSocketCloseException e) {
            logger.debug("Error while handling event", e);
            out.close(e.getWebSocketCloseStatus());
        } catch (Exception e) {
            logger.debug("Error while handling event", e);
            out.close(e);
        }
    }
Receives an event and returns a response if one should be sent.
graphql/src/main/java/com/linecorp/armeria/server/graphql/GraphqlWSSubProtocol.java