TXSynchronizationCommand extends BaseCommand { @Override public void cmdExecute(final Message clientMessage, final ServerConnection serverConnection, final SecurityService securityService, long start) throws IOException, ClassNotFoundException, InterruptedException { final boolean isDebugEnabled = logger.isDebugEnabled(); serverConnection.setAsTrue(REQUIRES_RESPONSE); CompletionType type = CompletionType.values()[clientMessage.getPart(0).getInt()]; clientMessage.getPart(1).getInt(); final Part statusPart; if (type == CompletionType.AFTER_COMPLETION) { statusPart = clientMessage.getPart(2); } else { statusPart = null; } final TXManagerImpl txMgr = getTXManager(serverConnection); final InternalDistributedMember member = getDistributedMember(serverConnection); final TXStateProxy txProxy = txMgr.getTXState(); assert txProxy != null; final TXId txId = txProxy.getTxId(); TXCommitMessage commitMessage = txMgr.getRecentlyCompletedMessage(txId); if (commitMessage != null && commitMessage != TXCommitMessage.ROLLBACK_MSG) { assert type == CompletionType.AFTER_COMPLETION; try { writeCommitResponse(clientMessage, serverConnection, commitMessage); } catch (IOException e) { if (isDebugEnabled) { logger.debug("Problem writing reply to client", e); } } catch (RuntimeException e) { try { writeException(clientMessage, e, false, serverConnection); } catch (IOException ioe) { if (isDebugEnabled) { logger.debug("Problem writing reply to client", ioe); } } } serverConnection.setAsTrue(RESPONDED); return; } try { if (type == CompletionType.BEFORE_COMPLETION) { if (isDebugEnabled) { logger.debug("Executing beforeCompletion() notification for transaction {}", clientMessage.getTransactionId()); } Throwable failureException = null; try { txProxy.setIsJTA(true); txProxy.beforeCompletion(); try { writeReply(clientMessage, serverConnection); } catch (IOException e) { if (isDebugEnabled) { logger.debug("Problem writing reply to client", e); } } serverConnection.setAsTrue(RESPONDED); } catch (ReplyException e) { failureException = e.getCause(); } catch (Exception e) { failureException = e; } if (failureException != null) { try { writeException(clientMessage, failureException, false, serverConnection); } catch (IOException ioe) { if (isDebugEnabled) { logger.debug("Problem writing reply to client", ioe); } } serverConnection.setAsTrue(RESPONDED); } } else { try { int status = statusPart.getInt(); if (isDebugEnabled) { logger.debug("Executing afterCompletion({}) notification for transaction {}", status, clientMessage.getTransactionId()); } txProxy.setIsJTA(true); txProxy.setCommitOnBehalfOfRemoteStub(true); txProxy.afterCompletion(status); TXCommitMessage cmsg = txProxy.getCommitMessage(); try { writeCommitResponse(clientMessage, serverConnection, cmsg); txMgr.removeHostedTXState(txProxy.getTxId()); } catch (IOException e) { if (isDebugEnabled || (e instanceof MessageTooLargeException)) { logger.warn("Problem writing reply to client", e); } } serverConnection.setAsTrue(RESPONDED); } catch (RuntimeException e) { try { writeException(clientMessage, e, false, serverConnection); } catch (IOException ioe) { if (isDebugEnabled) { logger.debug("Problem writing reply to client", ioe); } } serverConnection.setAsTrue(RESPONDED); } } } catch (Exception e) { writeException(clientMessage, MessageType.EXCEPTION, e, false, serverConnection); serverConnection.setAsTrue(RESPONDED); } if (isDebugEnabled) { logger.debug("Sent tx synchronization response"); } }  static Command getCommand(); @Override void cmdExecute(final Message clientMessage, final ServerConnection serverConnection,
      final SecurityService securityService, long start);  }
@Test public void commandCanInvokeBeforeCompletion() throws Exception { when(part0.getInt()).thenReturn(TXSynchronizationOp.CompletionType.BEFORE_COMPLETION.ordinal()); doCallRealMethod().when(command).cmdExecute(clientMessage, serverConnection, null, 1); command.cmdExecute(clientMessage, serverConnection, null, 1); verify(txStateProxy, times(1)).beforeCompletion(); verify(serverConnection, times(1)).setAsTrue(Command.RESPONDED); }