@Test public void testProcessRequest_Transaction() throws RemotingCommandException { brokerController.setTransactionalMessageService(transactionMsgService); when(brokerController.getTransactionalMessageService().asyncPrepareMessage(any(MessageExtBrokerInner.class))) .thenReturn(CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK)))); RemotingCommand request = createSendTransactionMsgCommand(RequestCode.SEND_MESSAGE); final RemotingCommand[] response = new RemotingCommand[1]; doAnswer(new Answer() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { response[0] = invocation.getArgument(0); return null; } }).when(handlerContext).writeAndFlush(any(Object.class)); RemotingCommand responseToReturn = sendMessageProcessor.processRequest(handlerContext, request); if (responseToReturn != null) { assertThat(response[0]).isNull(); response[0] = responseToReturn; } assertThat(response[0].getCode()).isEqualTo(ResponseCode.SUCCESS); }
@Override public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { RemotingCommand response = null; try { response = asyncProcessRequest(ctx, request).get(); } catch (InterruptedException | ExecutionException e) { log.error("process SendMessage error, request : " + request.toString(), e); } return response; }
SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor { @Override public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { RemotingCommand response = null; try { response = asyncProcessRequest(ctx, request).get(); } catch (InterruptedException | ExecutionException e) { log.error("process SendMessage error, request : " + request.toString(), e); } return response; } }
SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor { @Override public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { RemotingCommand response = null; try { response = asyncProcessRequest(ctx, request).get(); } catch (InterruptedException | ExecutionException e) { log.error("process SendMessage error, request : " + request.toString(), e); } return response; } SendMessageProcessor(final BrokerController brokerController); }
SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor { @Override public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { RemotingCommand response = null; try { response = asyncProcessRequest(ctx, request).get(); } catch (InterruptedException | ExecutionException e) { log.error("process SendMessage error, request : " + request.toString(), e); } return response; } SendMessageProcessor(final BrokerController brokerController); @Override RemotingCommand processRequest(ChannelHandlerContext ctx,
                                          RemotingCommand request); @Override void asyncProcessRequest(ChannelHandlerContext ctx, RemotingCommand request, RemotingResponseCallback responseCallback); CompletableFuture<RemotingCommand> asyncProcessRequest(ChannelHandlerContext ctx,
                                                                  RemotingCommand request); @Override boolean rejectRequest(); boolean hasConsumeMessageHook(); void executeConsumeMessageHookAfter(final ConsumeMessageContext context); SocketAddress getStoreHost(); void registerConsumeMessageHook(List<ConsumeMessageHook> consumeMessageHookList); }
SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor { @Override public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { RemotingCommand response = null; try { response = asyncProcessRequest(ctx, request).get(); } catch (InterruptedException | ExecutionException e) { log.error("process SendMessage error, request : " + request.toString(), e); } return response; } SendMessageProcessor(final BrokerController brokerController); @Override RemotingCommand processRequest(ChannelHandlerContext ctx,
                                          RemotingCommand request); @Override void asyncProcessRequest(ChannelHandlerContext ctx, RemotingCommand request, RemotingResponseCallback responseCallback); CompletableFuture<RemotingCommand> asyncProcessRequest(ChannelHandlerContext ctx,
                                                                  RemotingCommand request); @Override boolean rejectRequest(); boolean hasConsumeMessageHook(); void executeConsumeMessageHookAfter(final ConsumeMessageContext context); SocketAddress getStoreHost(); void registerConsumeMessageHook(List<ConsumeMessageHook> consumeMessageHookList);  }