CommandExecutorTask implements Runnable { @Override public void run() { RootInvokeHook rootInvokeHook = new SPIRootInvokeHook(); rootInvokeHook.start(); boolean isNeedFlush = false; int connectionSize = 0; try (PacketPayload payload = databaseProtocolFrontendEngine.getCodecEngine().createPacketPayload((ByteBuf) message)) { ConnectionStatus connectionStatus = backendConnection.getConnectionStatus(); if (!backendConnection.getTransactionStatus().isInConnectionHeldTransaction()) { connectionStatus.waitUntilConnectionRelease(); connectionStatus.switchToUsing(); } isNeedFlush = executeCommand(context, payload, backendConnection); connectionSize = backendConnection.getConnectionSize(); } catch (final Exception ex) { processException(ex); } finally { Collection<SQLException> exceptions = closeExecutionResources(); if (isNeedFlush) { context.flush(); } if (!backendConnection.getTransactionStatus().isInConnectionHeldTransaction()) { exceptions.addAll(backendConnection.closeConnections(false)); } processClosedExceptions(exceptions); rootInvokeHook.finish(connectionSize); } }  @Override void run();  }
@Test public void assertRunNeedFlushByFalse() throws SQLException { when(backendConnection.getConnectionSize()).thenReturn(1); when(queryCommandExecutor.execute()).thenReturn(Collections.emptyList()); when(executeEngine.getCommandPacket(eq(payload), eq(commandPacketType), eq(backendConnection))).thenReturn(commandPacket); when(executeEngine.getCommandExecutor(eq(commandPacketType), eq(commandPacket), eq(backendConnection))).thenReturn(queryCommandExecutor); when(executeEngine.getCommandPacketType(eq(payload))).thenReturn(commandPacketType); when(engine.getCommandExecuteEngine()).thenReturn(executeEngine); when(backendConnection.getConnectionStatus()).thenReturn(connectionStatus); when(codecEngine.createPacketPayload(eq(message))).thenReturn(payload); when(engine.getCodecEngine()).thenReturn(codecEngine); when(backendConnection.closeResultSets()).thenReturn(Collections.emptyList()); when(backendConnection.closeStatements()).thenReturn(Collections.emptyList()); when(backendConnection.closeConnections(false)).thenReturn(Collections.emptyList()); CommandExecutorTask actual = new CommandExecutorTask(engine, backendConnection, handlerContext, message); actual.run(); verify(connectionStatus).waitUntilConnectionRelease(); verify(connectionStatus).switchToUsing(); }
@Test public void assertRunNeedFlushByTrue() throws SQLException { when(backendConnection.getConnectionSize()).thenReturn(1); when(queryCommandExecutor.execute()).thenReturn(Collections.singletonList(databasePacket)); when(executeEngine.getCommandPacket(eq(payload), eq(commandPacketType), eq(backendConnection))).thenReturn(commandPacket); when(executeEngine.getCommandExecutor(eq(commandPacketType), eq(commandPacket), eq(backendConnection))).thenReturn(queryCommandExecutor); when(executeEngine.getCommandPacketType(eq(payload))).thenReturn(commandPacketType); when(engine.getCommandExecuteEngine()).thenReturn(executeEngine); when(backendConnection.getConnectionStatus()).thenReturn(connectionStatus); when(codecEngine.createPacketPayload(eq(message))).thenReturn(payload); when(engine.getCodecEngine()).thenReturn(codecEngine); when(backendConnection.closeResultSets()).thenReturn(Collections.emptyList()); when(backendConnection.closeStatements()).thenReturn(Collections.emptyList()); when(backendConnection.closeConnections(false)).thenReturn(Collections.emptyList()); CommandExecutorTask actual = new CommandExecutorTask(engine, backendConnection, handlerContext, message); actual.run(); verify(connectionStatus).waitUntilConnectionRelease(); verify(connectionStatus).switchToUsing(); verify(handlerContext).write(databasePacket); verify(handlerContext).flush(); verify(executeEngine).writeQueryData(handlerContext, backendConnection, queryCommandExecutor, 1); }
@Test public void assertRunByCommandExecutor() throws SQLException { when(frontendContext.isFlushForPerCommandPacket()).thenReturn(true); when(engine.getFrontendContext()).thenReturn(frontendContext); when(backendConnection.getConnectionSize()).thenReturn(1); when(commandExecutor.execute()).thenReturn(Collections.singletonList(databasePacket)); when(executeEngine.getCommandPacket(eq(payload), eq(commandPacketType), eq(backendConnection))).thenReturn(commandPacket); when(executeEngine.getCommandExecutor(eq(commandPacketType), eq(commandPacket), eq(backendConnection))).thenReturn(commandExecutor); when(executeEngine.getCommandPacketType(eq(payload))).thenReturn(commandPacketType); when(engine.getCommandExecuteEngine()).thenReturn(executeEngine); when(backendConnection.getConnectionStatus()).thenReturn(connectionStatus); when(codecEngine.createPacketPayload(eq(message))).thenReturn(payload); when(engine.getCodecEngine()).thenReturn(codecEngine); when(backendConnection.closeResultSets()).thenReturn(Collections.emptyList()); when(backendConnection.closeStatements()).thenReturn(Collections.emptyList()); when(backendConnection.closeConnections(false)).thenReturn(Collections.emptyList()); CommandExecutorTask actual = new CommandExecutorTask(engine, backendConnection, handlerContext, message); actual.run(); verify(connectionStatus).waitUntilConnectionRelease(); verify(connectionStatus).switchToUsing(); verify(handlerContext).write(databasePacket); verify(handlerContext).flush(); }
@Test public void assertRunWithError() { RuntimeException mockException = new RuntimeException("mock"); when(backendConnection.getConnectionStatus()).thenThrow(mockException); when(codecEngine.createPacketPayload(message)).thenReturn(payload); when(engine.getCodecEngine()).thenReturn(codecEngine); when(executeEngine.getErrorPacket(eq(mockException))).thenReturn(databasePacket); when(executeEngine.getOtherPacket()).thenReturn(Optional.of(databasePacket)); when(engine.getCommandExecuteEngine()).thenReturn(executeEngine); when(backendConnection.closeResultSets()).thenReturn(Collections.emptyList()); when(backendConnection.closeStatements()).thenReturn(Collections.emptyList()); when(backendConnection.closeConnections(false)).thenReturn(Collections.emptyList()); CommandExecutorTask actual = new CommandExecutorTask(engine, backendConnection, handlerContext, message); actual.run(); verify(handlerContext, atLeast(2)).writeAndFlush(databasePacket); }