AbstractEvalOpProcessor extends AbstractOpProcessor { protected void evalOpInternal(final Context ctx, final Supplier<GremlinExecutor> gremlinExecutorSupplier, final BindingSupplier bindingsSupplier) throws OpProcessorException { final Timer.Context timerContext = evalOpTimer.time(); final RequestMessage msg = ctx.getRequestMessage(); final GremlinExecutor gremlinExecutor = gremlinExecutorSupplier.get(); final Settings settings = ctx.getSettings(); final Map<String, Object> args = msg.getArgs(); final String script = (String) args.get(Tokens.ARGS_GREMLIN); final String language = args.containsKey(Tokens.ARGS_LANGUAGE) ? (String) args.get(Tokens.ARGS_LANGUAGE) : null; final Bindings bindings = new SimpleBindings(); final boolean managedTransactionsForRequest = manageTransactions ? true : (Boolean) args.getOrDefault(Tokens.ARGS_MANAGE_TRANSACTION, false); final long seto = args.containsKey(Tokens.ARGS_EVAL_TIMEOUT) ? ((Number) args.get(Tokens.ARGS_EVAL_TIMEOUT)).longValue() : settings.getEvaluationTimeout(); final GremlinExecutor.LifeCycle lifeCycle = GremlinExecutor.LifeCycle.build() .evaluationTimeoutOverride(seto) .afterFailure((b,t) -> { if (managedTransactionsForRequest) attemptRollback(msg, ctx.getGraphManager(), settings.strictTransactionManagement); }) .beforeEval(b -> { try { b.putAll(bindingsSupplier.get()); } catch (OpProcessorException ope) { throw new RuntimeException(ope); } }) .withResult(o -> { final Iterator itty = IteratorUtils.asIterator(o); logger.debug("Preparing to iterate results from - {} - in thread [{}]", msg, Thread.currentThread().getName()); if (settings.authentication.enableAuditLog) { String address = ctx.getChannelHandlerContext().channel().remoteAddress().toString(); if (address.startsWith("/") && address.length() > 1) address = address.substring(1); auditLogger.info("User with address {} requested: {}", address, script); } try { handleIterator(ctx, itty); } catch (Exception ex) { if (managedTransactionsForRequest) attemptRollback(msg, ctx.getGraphManager(), settings.strictTransactionManagement); throw new RuntimeException(ex); } }).create(); final CompletableFuture<Object> evalFuture = gremlinExecutor.eval(script, language, bindings, lifeCycle); evalFuture.handle((v, t) -> { timerContext.stop(); if (t != null) { if (t instanceof OpProcessorException) { ctx.writeAndFlush(((OpProcessorException) t).getResponseMessage()); } else if (t instanceof TimedInterruptTimeoutException) { final String errorMessage = String.format("A timeout occurred within the script during evaluation of [%s] - consider increasing the limit given to TimedInterruptCustomizerProvider", msg); logger.warn(errorMessage); ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT) .statusMessage("Timeout during script evaluation triggered by TimedInterruptCustomizerProvider") .statusAttributeException(t).create()); } else if (t instanceof TimeoutException) { final String errorMessage = String.format("Script evaluation exceeded the configured threshold for request [%s]", msg); logger.warn(errorMessage, t); ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT) .statusMessage(t.getMessage()) .statusAttributeException(t).create()); } else { if (t instanceof MultipleCompilationErrorsException && t.getMessage().contains("Method too large") && ((MultipleCompilationErrorsException) t).getErrorCollector().getErrorCount() == 1) { final String errorMessage = String.format("The Gremlin statement that was submitted exceeds the maximum compilation size allowed by the JVM, please split it into multiple smaller statements - %s", trimMessage(msg)); logger.warn(errorMessage); ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_EVALUATION) .statusMessage(errorMessage) .statusAttributeException(t).create()); } else { final String errorMessage = (t.getMessage() == null) ? t.toString() : t.getMessage(); logger.warn(String.format("Exception processing a script on request [%s].", msg), t); ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_EVALUATION) .statusMessage(errorMessage) .statusAttributeException(t).create()); } } } return null; }); } protected  AbstractEvalOpProcessor(final boolean manageTransactions); abstract ThrowingConsumer<Context> getEvalOp(); abstract Optional<ThrowingConsumer<Context>> selectOther(final RequestMessage requestMessage); @Override ThrowingConsumer<Context> select(final Context ctx); static final Timer evalOpTimer; static final String CONFIG_MAX_PARAMETERS; static final int DEFAULT_MAX_PARAMETERS; }
@Test public void evalOpInternalShouldHandleAllEvaluationExceptions() throws OpProcessorException { final AbstractEvalOpProcessor processor = new StandardOpProcessor(); final RequestMessage request = RequestMessage.build("test"). addArg(Tokens.ARGS_LANGUAGE, "gremlin-groovy"). addArg(Tokens.ARGS_GREMLIN, "1+1").create(); final Settings settings = new Settings(); final ChannelHandlerContext ctx = Mockito.mock(ChannelHandlerContext.class); final ArgumentCaptor<ResponseMessage> responseCaptor = ArgumentCaptor.forClass(ResponseMessage.class); final GremlinExecutor gremlinExecutor = Mockito.mock(GremlinExecutor.class); final CompletableFuture<Object> exFut = new CompletableFuture<>(); Mockito.when(gremlinExecutor.eval(anyString(), anyString(), Mockito.any(), Mockito.<GremlinExecutor.LifeCycle>any())) .thenReturn(exFut); exFut.completeExceptionally(new IllegalStateException("test-exception")); final Context contextspy = Mockito.spy(new Context(request, ctx, settings, null, gremlinExecutor, null)); processor.evalOpInternal(contextspy, contextspy::getGremlinExecutor, SimpleBindings::new); Mockito.verify(contextspy, Mockito.times(1)).writeAndFlush(responseCaptor.capture()); assertEquals(ResponseStatusCode.SERVER_ERROR_EVALUATION, responseCaptor.getValue().getStatus().getCode()); assertEquals(request.getRequestId(), responseCaptor.getValue().getRequestId()); assertThat(responseCaptor.getValue().getStatus().getMessage(), CoreMatchers.containsString("test-exception")); }