@Test public void onNext() throws InterruptedException, ExecutionException, TimeoutException { doAnswer(invocation -> { String aggregateId = invocation.getArgument(0); Consumer<SerializedEvent> consumer = invocation.getArgument(3); for (int i = 0; i < 10; i++) { Event event = Event.newBuilder().setAggregateSequenceNumber(i).setAggregateIdentifier(aggregateId) .build(); consumer.accept(new SerializedEvent(event)); } return null; }).when(aggregateReader).readEvents(anyString(), anyBoolean(), anyLong(), any(Consumer.class)); testSubject.onNext(QueryEventsRequest.newBuilder() .setQuery("aggregateIdentifier = \"12345\"") .setNumberOfPermits(1000) .build()); List<QueryEventsResponse> responses = completableResult.get(2, TimeUnit.SECONDS); assertEquals(12, responses.size()); }
@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); } }
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); } } }
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); } } QueryEventsRequestStreamObserver(EventWriteStorage eventWriteStorage, EventStreamReader eventStreamReader,
                                            AggregateReader aggregateReader,
                                            long defaultLimit, long timeout, EventDecorator eventDecorator,
                                            StreamObserver<QueryEventsResponse> responseObserver); }
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); } } QueryEventsRequestStreamObserver(EventWriteStorage eventWriteStorage, EventStreamReader eventStreamReader,
                                            AggregateReader aggregateReader,
                                            long defaultLimit, long timeout, EventDecorator eventDecorator,
                                            StreamObserver<QueryEventsResponse> responseObserver); @Override void onNext(QueryEventsRequest queryEventsRequest); @Override void onError(Throwable throwable); @Override void onCompleted(); }
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); } } QueryEventsRequestStreamObserver(EventWriteStorage eventWriteStorage, EventStreamReader eventStreamReader,
                                            AggregateReader aggregateReader,
                                            long defaultLimit, long timeout, EventDecorator eventDecorator,
                                            StreamObserver<QueryEventsResponse> responseObserver); @Override void onNext(QueryEventsRequest queryEventsRequest); @Override void onError(Throwable throwable); @Override void onCompleted(); static final int TIME_WINDOW_FIELD; static final String TIME_WINDOW_CUSTOM; static final String COLUMN_NAME_TOKEN; }