class QueryEventsRequestStreamObserver implements StreamObserver<QueryEventsRequest> {

    @Override
    public void onNext(QueryEventsRequest queryEventsRequest) {
        try {
            Sender sender = senderRef.updateAndGet(s -> {
                if (s == null) {
                    return new Sender(queryEventsRequest.getNumberOfPermits(),
                                      queryEventsRequest.getLiveEvents(),
                                      System.currentTimeMillis() + timeout);
                }
                return s;
            });
            if (sender.start()) {
                long connectionToken = eventStreamReader.getLastToken();
                long maxToken = Long.MAX_VALUE;
                long minConnectionToken = StringUtils.isEmpty(queryEventsRequest.getQuery()) ? Math.max(
                        connectionToken - defaultLimit, 0) : 0;
                String queryString = StringUtils.isEmpty(queryEventsRequest.getQuery()) ?
                        "token >= " + minConnectionToken + "| sortby(token)" : queryEventsRequest.getQuery();
                String timeWindow = getTimeWindow(queryEventsRequest);
                if (!TIME_WINDOW_CUSTOM.equals(timeWindow)) {
                    queryString = queryString + " | " + timeWindow;
                }

                Query query = new EventStoreQueryParser().parse(queryString);
                query.addDefaultLimit(defaultLimit);
                String aggregateIdentifier = null;
                for (int i = 0; i < query.size(); i++) {
                    if (query.get(i) instanceof FunctionExpr) {
                        FunctionExpr functionExpr = (FunctionExpr) query.get(i);
                        if (isProjectOperation(functionExpr.operator())) {
                            break;
                        }
                        if (COLUMN_NAME_TOKEN.equals(functionExpr.getParameters().get(0).getLiteral()) &&
                                operatorIn(functionExpr.operator(), "=", ">", ">=")) {
                            minConnectionToken = getValueAsLong(minConnectionToken,
                                                                functionExpr.getParameters().get(1));
                        }
                        if (COLUMN_NAME_TOKEN.equals(functionExpr.getParameters().get(0).getLiteral()) &&
                                operatorIn(functionExpr.operator(), "<", "<=", "=")) {
                            maxToken = getValueAsLong(maxToken, functionExpr.getParameters().get(1));
                        }
                        if ("aggregateIdentifier".equals(functionExpr.getParameters().get(0).getLiteral()) &&
                                "=".equals(functionExpr.operator())) {
                            aggregateIdentifier = functionExpr.getParameters().get(1).getLiteral();
                        }
                    }
                }
                logger.info("Executing query: {}", query);
                pipeLine = new QueryProcessor().buildPipeline(query, this::send);
                sendColumns(pipeLine, aggregateIdentifier != null);
                if (queryEventsRequest.getLiveEvents() && maxToken > connectionToken) {
                    registration = eventWriteStorage.registerEventListener((token, events) -> pushEventFromStream(token,
                                                                                                                  events,
                                                                                                                  pipeLine));
                }
                if (aggregateIdentifier == null) {
                    QueryOptions queryOptions = new QueryOptions(minConnectionToken, maxToken, query.getStartTime());
                    senderService.submit(() -> {
                        eventStreamReader.query(queryOptions,
                                                event -> pushEvent(event, pipeLine));
                        Optional.ofNullable(senderRef.get())
                                .ifPresent(Sender::completed);
                    });
                } else {
                    String finalAggregateIdentifier = aggregateIdentifier;
                    senderService.submit(() -> {
                        aggregateReader.readEvents(finalAggregateIdentifier, false, 0, serializedEvent -> {
                            Event event = serializedEvent.asEvent();
                            if (event.getTimestamp() >= query.getStartTime()) {
                                pushEvent(EventWithToken.newBuilder()
                                                        .setEvent(event)
                                                        .setToken(event.getAggregateSequenceNumber())
                                                        .build(), pipeLine);
                            }
                        });
                        Optional.ofNullable(senderRef.get())
                                .ifPresent(Sender::completed);
                    });
                }
            } else {
                Optional.ofNullable(senderRef.get())
                        .ifPresent(s -> s.addPermits(queryEventsRequest.getNumberOfPermits()));
            }
        } catch (Exception pe) {
            responseObserver.onError(pe);
            senderRef.set(null);
        }
    }

    public  QueryEventsRequestStreamObserver(EventWriteStorage eventWriteStorage, EventStreamReader eventStreamReader,
                                            AggregateReader aggregateReader,
                                            long defaultLimit, long timeout, EventDecorator eventDecorator,
                                            StreamObserver<QueryEventsResponse> responseObserver);

}