ServiceProxy extends TOMSender { @Override public void replyReceived(TOMMessage reply) { Logger.println("Synchronously received reply from " + reply.getSender() + " with sequence number " + reply.getSequence()); try { canReceiveLock.lock(); if (reqId == -1) { Logger.println("throwing out request: sender=" + reply.getSender() + " reqId=" + reply.getSequence()); canReceiveLock.unlock(); return; } int pos = getViewManager().getCurrentViewPos(reply.getSender()); if (pos < 0) { canReceiveLock.unlock(); return; } int sameContent = 1; if (reply.getSequence() == reqId && reply.getReqType() == requestType) { Logger.println("Receiving reply from " + reply.getSender() + " with reqId:" + reply.getSequence() + ". Putting on pos=" + pos); if (requestType == TOMMessageType.UNORDERED_HASHED_REQUEST) { response = hashResponseController.getResponse(pos, reply); if (response != null) { reqId = -1; this.sm.release(); canReceiveLock.unlock(); return; } } else { if (replies[pos] == null) { receivedReplies++; } replies[pos] = reply; for (int i = 0; i < replies.length; i++) { if ((i != pos || getViewManager().getCurrentViewN() == 1) && replies[i] != null && (comparator.compare(replies[i].getContent(), reply.getContent()) == 0)) { sameContent++; if (sameContent >= replyQuorum) { response = extractor.extractResponse(replies, sameContent, pos); reqId = -1; this.sm.release(); canReceiveLock.unlock(); return; } } } } if (response == null) { if (requestType.equals(TOMMessageType.ORDERED_REQUEST)) { if (receivedReplies == getViewManager().getCurrentViewN()) { reqId = -1; this.sm.release(); } } else if (requestType.equals(TOMMessageType.UNORDERED_HASHED_REQUEST)) { if (hashResponseController.getNumberReplies() == getViewManager().getCurrentViewN()) { reqId = -1; this.sm.release(); } } else { if (receivedReplies != sameContent) { reqId = -1; this.sm.release(); } } } } else { Logger.println("Ignoring reply from " + reply.getSender() + " with reqId:" + reply.getSequence() + ". Currently wait reqId= " + reqId); } canReceiveLock.unlock(); } catch (Exception ex) { System.out.println("Problem at ServiceProxy.ReplyReceived()"); ex.printStackTrace(); canReceiveLock.unlock(); } } ServiceProxy(int processId);  ServiceProxy(int processId, String configHome);  ServiceProxy(int processId, String configHome, Comparator<byte[]> replyComparator,
			Extractor replyExtractor); int getInvokeTimeout(); int getInvokeUnorderedHashedTimeout(); void setInvokeTimeout(int invokeTimeout); void setInvokeUnorderedHashedTimeout(int timeout); byte[] invokeOrdered(byte[] request); byte[] invokeUnordered(byte[] request); byte[] invokeUnorderedHashed(byte[] request); byte[] invoke(byte[] request, TOMMessageType reqType); @Override void replyReceived(TOMMessage reply);  }
@Test public void testReplyReceived() { Field response; Field receivedReplies; Field reqId; Field replyQuorum; Field replies; Field requestType; try { response = ServiceProxy.class.getDeclaredField("response"); response.setAccessible(true); receivedReplies = ServiceProxy.class.getDeclaredField("receivedReplies"); receivedReplies.setAccessible(true); reqId = ServiceProxy.class.getDeclaredField("reqId"); reqId.setAccessible(true); replyQuorum = ServiceProxy.class.getDeclaredField("replyQuorum"); replyQuorum.setAccessible(true); replies = ServiceProxy.class.getDeclaredField("replies"); replies.setAccessible(true); requestType = ServiceProxy.class.getDeclaredField("requestType"); requestType.setAccessible(true); try { TOMMessage[] initReplies = new TOMMessage[4]; receivedReplies.set(proxy, 0); reqId.setInt(proxy, 1); replyQuorum.setInt(proxy, 3); replies.set(proxy, initReplies); requestType.set(proxy, TOMMessageType.ORDERED_REQUEST); response.set(proxy, null); TOMMessage msg = new TOMMessage(0, 1, 1, "response1".getBytes(), 0, TOMMessageType.ORDERED_REQUEST); proxy.replyReceived(msg); msg = new TOMMessage(1, 1, 1, "response1".getBytes(), 0, TOMMessageType.ORDERED_REQUEST); proxy.replyReceived(msg); msg = new TOMMessage(2, 1, 1, "response1".getBytes(), 0, TOMMessageType.ORDERED_REQUEST); proxy.replyReceived(msg); TOMMessage reply = (TOMMessage) response.get(proxy); Assert.assertEquals("response1", new String(reply.getContent())); initReplies = new TOMMessage[4]; receivedReplies.set(proxy, 0); reqId.setInt(proxy, 1); replyQuorum.setInt(proxy, 3); replies.set(proxy, initReplies); requestType.set(proxy, TOMMessageType.ORDERED_REQUEST); response.set(proxy, null); msg = new TOMMessage(0, 1, 1, "response1".getBytes(), 0, TOMMessageType.ORDERED_REQUEST); proxy.replyReceived(msg); msg = new TOMMessage(1, 1, 1, "response1".getBytes(), 0, TOMMessageType.ORDERED_REQUEST); proxy.replyReceived(msg); msg = new TOMMessage(2, 1, 1, "response2".getBytes(), 0, TOMMessageType.ORDERED_REQUEST); proxy.replyReceived(msg); msg = new TOMMessage(3, 1, 1, "response1".getBytes(), 0, TOMMessageType.ORDERED_REQUEST); proxy.replyReceived(msg); reply = (TOMMessage) response.get(proxy); Assert.assertEquals("response1", new String(reply.getContent())); initReplies = new TOMMessage[4]; receivedReplies.set(proxy, 0); reqId.setInt(proxy, 1); replyQuorum.setInt(proxy, 3); replies.set(proxy, initReplies); requestType.set(proxy, TOMMessageType.ORDERED_REQUEST); response.set(proxy, null); msg = new TOMMessage(0, 1, 1, "response1".getBytes(), 0, TOMMessageType.ORDERED_REQUEST); proxy.replyReceived(msg); msg = new TOMMessage(1, 1, 1, "response1".getBytes(), 0, TOMMessageType.ORDERED_REQUEST); proxy.replyReceived(msg); msg = new TOMMessage(2, 1, 1, "response2".getBytes(), 0, TOMMessageType.ORDERED_REQUEST); proxy.replyReceived(msg); msg = new TOMMessage(3, 1, 1, "response2".getBytes(), 0, TOMMessageType.ORDERED_REQUEST); proxy.replyReceived(msg); reply = (TOMMessage) response.get(proxy); Assert.assertNull(reply); Assert.assertEquals(-1, reqId.get(proxy)); initReplies = new TOMMessage[4]; receivedReplies.set(proxy, 0); reqId.setInt(proxy, 1); replies.set(proxy, initReplies); requestType.set(proxy, TOMMessageType.UNORDERED_REQUEST); response.set(proxy, null); msg = new TOMMessage(0, 1, 1, "response1".getBytes(), 0, TOMMessageType.UNORDERED_REQUEST); proxy.replyReceived(msg); msg = new TOMMessage(1, 1, 1, "response1".getBytes(), 0, TOMMessageType.UNORDERED_REQUEST); proxy.replyReceived(msg); msg = new TOMMessage(2, 1, 1, "response1".getBytes(), 0, TOMMessageType.UNORDERED_REQUEST); proxy.replyReceived(msg); reply = (TOMMessage) response.get(proxy); Assert.assertEquals("response1", new String(reply.getContent())); receivedReplies.set(proxy, 0); reqId.setInt(proxy, 1); replies.set(proxy, initReplies); requestType.set(proxy, TOMMessageType.UNORDERED_REQUEST); response.set(proxy, null); msg = new TOMMessage(0, 1, 1, "response1".getBytes(), 0, TOMMessageType.UNORDERED_REQUEST); proxy.replyReceived(msg); msg = new TOMMessage(1, 1, 1, "response2".getBytes(), 0, TOMMessageType.UNORDERED_REQUEST); proxy.replyReceived(msg); reply = (TOMMessage) response.get(proxy); Assert.assertEquals(-1, reqId.get(proxy)); } catch (IllegalArgumentException | IllegalAccessException e) { e.printStackTrace(); } } catch (NoSuchFieldException e) { e.printStackTrace(); } catch (SecurityException e) { e.printStackTrace(); } }