@Test public void testSocketRequestResponseChannelTest() { try { SocketRequestResponseChannel channel = new SocketRequestResponseChannel(2, 10); Integer key = new Integer(5); String connectionId = "test_connectionId"; ByteBuf buffer = PooledByteBufAllocator.DEFAULT.heapBuffer(1000); byte[] content = new byte[1000]; new Random().nextBytes(content); buffer.writeBytes(content); SocketServerRequest request = new SocketServerRequest(0, connectionId, buffer); channel.sendRequest(request); request = (SocketServerRequest) channel.receiveRequest(); Assert.assertEquals(request.getProcessor(), 0); Assert.assertEquals(request.getConnectionId(), connectionId); InputStream stream = request.getInputStream(); for (int i = 0; i < 1000; i++) { Assert.assertEquals((byte) stream.read(), content[i]); } request.release(); ResponseListenerMock mock = new ResponseListenerMock(); channel.addResponseListener(mock); MockSend mocksend = new MockSend(); channel.sendResponse(mocksend, request, null); Assert.assertEquals(mock.call, 1); SocketServerResponse response = (SocketServerResponse) channel.receiveResponse(0); Assert.assertEquals(response.getProcessor(), 0); } catch (Exception e) { Assert.assertEquals(true, false); } }
public SocketRequestResponseChannel(int numProcessors, int queueSize) { this.numProcessors = numProcessors; this.queueSize = queueSize; this.requestQueue = new ArrayBlockingQueue<>(this.queueSize); responseQueues = new ArrayList<>(this.numProcessors); responseListeners = new ArrayList<ResponseListener>(); for (int i = 0; i < this.numProcessors; i++) { responseQueues.add(i, new LinkedBlockingQueue<>()); } }
SocketRequestResponseChannel implements RequestResponseChannel { public SocketRequestResponseChannel(int numProcessors, int queueSize) { this.numProcessors = numProcessors; this.queueSize = queueSize; this.requestQueue = new ArrayBlockingQueue<>(this.queueSize); responseQueues = new ArrayList<>(this.numProcessors); responseListeners = new ArrayList<ResponseListener>(); for (int i = 0; i < this.numProcessors; i++) { responseQueues.add(i, new LinkedBlockingQueue<>()); } } }
SocketRequestResponseChannel implements RequestResponseChannel { public SocketRequestResponseChannel(int numProcessors, int queueSize) { this.numProcessors = numProcessors; this.queueSize = queueSize; this.requestQueue = new ArrayBlockingQueue<>(this.queueSize); responseQueues = new ArrayList<>(this.numProcessors); responseListeners = new ArrayList<ResponseListener>(); for (int i = 0; i < this.numProcessors; i++) { responseQueues.add(i, new LinkedBlockingQueue<>()); } } SocketRequestResponseChannel(int numProcessors, int queueSize); }
SocketRequestResponseChannel implements RequestResponseChannel { public SocketRequestResponseChannel(int numProcessors, int queueSize) { this.numProcessors = numProcessors; this.queueSize = queueSize; this.requestQueue = new ArrayBlockingQueue<>(this.queueSize); responseQueues = new ArrayList<>(this.numProcessors); responseListeners = new ArrayList<ResponseListener>(); for (int i = 0; i < this.numProcessors; i++) { responseQueues.add(i, new LinkedBlockingQueue<>()); } } SocketRequestResponseChannel(int numProcessors, int queueSize); @Override void sendRequest(NetworkRequest request); @Override void sendResponse(Send payloadToSend, NetworkRequest originalRequest, ServerNetworkResponseMetrics metrics); @Override void closeConnection(NetworkRequest originalRequest); @Override NetworkRequest receiveRequest(); NetworkResponse receiveResponse(int processor); void addResponseListener(ResponseListener listener); int getRequestQueueSize(); int getResponseQueueSize(int processor); int getNumberOfProcessors(); void shutdown(); }
SocketRequestResponseChannel implements RequestResponseChannel { public SocketRequestResponseChannel(int numProcessors, int queueSize) { this.numProcessors = numProcessors; this.queueSize = queueSize; this.requestQueue = new ArrayBlockingQueue<>(this.queueSize); responseQueues = new ArrayList<>(this.numProcessors); responseListeners = new ArrayList<ResponseListener>(); for (int i = 0; i < this.numProcessors; i++) { responseQueues.add(i, new LinkedBlockingQueue<>()); } } SocketRequestResponseChannel(int numProcessors, int queueSize); @Override void sendRequest(NetworkRequest request); @Override void sendResponse(Send payloadToSend, NetworkRequest originalRequest, ServerNetworkResponseMetrics metrics); @Override void closeConnection(NetworkRequest originalRequest); @Override NetworkRequest receiveRequest(); NetworkResponse receiveResponse(int processor); void addResponseListener(ResponseListener listener); int getRequestQueueSize(); int getResponseQueueSize(int processor); int getNumberOfProcessors(); void shutdown();  }