@Test public void shouldProcessMessagesInCorrectOrder() { messageQueue.addMessage(stubPeer, MESSAGE_ID, 2, stubContent); messageQueue.addMessage(stubPeer, OTHER_MESSAGE_ID, 1, stubContent); InOrder inOrder = Mockito.inOrder(processorsService); inOrder.verify(processorsService).getMessageProcessor(OTHER_MESSAGE_ID); inOrder.verify(processorsService).getMessageProcessor(MESSAGE_ID); }
public synchronized void addMessage(Peer sourcePeer, int messageType, int dependencyId, ByteBuf content) { if (checkDependency(dependencyId)) { byte[] contentBytes = new byte[content.readableBytes()]; content.readBytes(contentBytes); messageQueue.add(new IncomingEnvelope(sourcePeer, messageType, dependencyId, new SerializationBuffer(contentBytes))); } processIncomingMessages(); }
IncomingMessageQueue { public synchronized void addMessage(Peer sourcePeer, int messageType, int dependencyId, ByteBuf content) { if (checkDependency(dependencyId)) { byte[] contentBytes = new byte[content.readableBytes()]; content.readBytes(contentBytes); messageQueue.add(new IncomingEnvelope(sourcePeer, messageType, dependencyId, new SerializationBuffer(contentBytes))); } processIncomingMessages(); } }
IncomingMessageQueue { public synchronized void addMessage(Peer sourcePeer, int messageType, int dependencyId, ByteBuf content) { if (checkDependency(dependencyId)) { byte[] contentBytes = new byte[content.readableBytes()]; content.readBytes(contentBytes); messageQueue.add(new IncomingEnvelope(sourcePeer, messageType, dependencyId, new SerializationBuffer(contentBytes))); } processIncomingMessages(); } IncomingMessageQueue(MessageProcessorsService messageProcessorsService); }
IncomingMessageQueue { public synchronized void addMessage(Peer sourcePeer, int messageType, int dependencyId, ByteBuf content) { if (checkDependency(dependencyId)) { byte[] contentBytes = new byte[content.readableBytes()]; content.readBytes(contentBytes); messageQueue.add(new IncomingEnvelope(sourcePeer, messageType, dependencyId, new SerializationBuffer(contentBytes))); } processIncomingMessages(); } IncomingMessageQueue(MessageProcessorsService messageProcessorsService); synchronized void addMessage(Peer sourcePeer, int messageType, int dependencyId, ByteBuf content); }
IncomingMessageQueue { public synchronized void addMessage(Peer sourcePeer, int messageType, int dependencyId, ByteBuf content) { if (checkDependency(dependencyId)) { byte[] contentBytes = new byte[content.readableBytes()]; content.readBytes(contentBytes); messageQueue.add(new IncomingEnvelope(sourcePeer, messageType, dependencyId, new SerializationBuffer(contentBytes))); } processIncomingMessages(); } IncomingMessageQueue(MessageProcessorsService messageProcessorsService); synchronized void addMessage(Peer sourcePeer, int messageType, int dependencyId, ByteBuf content);  }