@Test public void testProcessUpdate() throws Exception { List<SxpBindingFields> add = new ArrayList<>(), dell = new ArrayList<>(); List<Peer> peerList = new ArrayList<>(); peerList.add(getPeer("127.0.0.0", 0)); peerList.add(getPeer("1.1.1.1", 1)); peerList.add(getPeer("127.0.0.2", 2)); add.add(getBinding("5.5.5.5/32", 20, new PeerSequenceBuilder().setPeer(new ArrayList<>()).build())); add.add(getBinding("5.5.5.5/32", 25, new PeerSequenceBuilder().setPeer(new ArrayList<>()).build())); add.add(getBinding("15.5.15.0/24", 40, new PeerSequenceBuilder().setPeer(new ArrayList<>()).build())); add.add(getBinding("5.0.5.50/32", 120, new PeerSequenceBuilder().setPeer(peerList).build())); for (int i = 0; i < 5; i++) handler.processUpdate(dell, add, connection); taskCaptor.getValue().call(); assertDatabase(sxpDatabaseInf.getBindings(), getIpPrefixes("5.5.5.5/32", "15.5.15.0/24")); assertDatabase(masterDatabaseInf.getBindings(), getIpPrefixes("5.5.5.5/32", "15.5.15.0/24")); add.clear(); dell.clear(); when(connection.getId()).thenReturn(NodeId.getDefaultInstance("0.0.0.1")); dell.add(getBinding("5.5.5.5/32", 20, new PeerSequenceBuilder().setPeer(new ArrayList<>()).build())); dell.add(getBinding("15.5.15.0/24", 30, new PeerSequenceBuilder().setPeer(new ArrayList<>()).build())); add.add(getBinding("55.2.0.0/16", 10, new PeerSequenceBuilder().setPeer(new ArrayList<>()).build())); add.add(getBinding("5.5.0.0/32", 80, new PeerSequenceBuilder().setPeer(new ArrayList<>()).build())); for (int i = 0; i < 5; i++) handler.processUpdate(dell, add, connection); taskCaptor.getValue().call(); assertDatabase(masterDatabaseInf.getBindings(), getIpPrefixes("5.5.5.5/32", "55.2.0.0/16", "5.5.0.0/32", "15.5.15.0/24")); add.clear(); dell.clear(); dell.add(getBinding("55.2.0.0/16", 10, new PeerSequenceBuilder().setPeer(new ArrayList<>()).build())); dell.add(getBinding("5.5.0.0/32", 80, new PeerSequenceBuilder().setPeer(new ArrayList<>()).build())); for (int i = 0; i < 5; i++) handler.processUpdate(dell, add, connection); taskCaptor.getValue().call(); assertDatabase(masterDatabaseInf.getBindings(), getIpPrefixes("5.5.5.5/32", "15.5.15.0/24")); }
public <T extends SxpBindingFields> void processUpdate(final List<T> databaseDelete, final List<T> databaseAdd, final SxpConnection connection) { final Deque<DecodedMessage> deque; final DecodedMessage message = new DecodedMessage(databaseDelete, databaseAdd); synchronized (buffer) { if (!buffer.containsKey(connection)) { buffer.put(connection, (deque = new ArrayDeque<>())); } else { deque = buffer.get(connection); } } deque.addLast(message); if (deque.size() == 1) { connection.getOwner() .getWorker() .executeTaskInSequence(() -> updateMessageCallback(message, connection, deque), ThreadsWorker.WorkerType.INBOUND, connection); } }
BindingHandler { public <T extends SxpBindingFields> void processUpdate(final List<T> databaseDelete, final List<T> databaseAdd, final SxpConnection connection) { final Deque<DecodedMessage> deque; final DecodedMessage message = new DecodedMessage(databaseDelete, databaseAdd); synchronized (buffer) { if (!buffer.containsKey(connection)) { buffer.put(connection, (deque = new ArrayDeque<>())); } else { deque = buffer.get(connection); } } deque.addLast(message); if (deque.size() == 1) { connection.getOwner() .getWorker() .executeTaskInSequence(() -> updateMessageCallback(message, connection, deque), ThreadsWorker.WorkerType.INBOUND, connection); } } }
BindingHandler { public <T extends SxpBindingFields> void processUpdate(final List<T> databaseDelete, final List<T> databaseAdd, final SxpConnection connection) { final Deque<DecodedMessage> deque; final DecodedMessage message = new DecodedMessage(databaseDelete, databaseAdd); synchronized (buffer) { if (!buffer.containsKey(connection)) { buffer.put(connection, (deque = new ArrayDeque<>())); } else { deque = buffer.get(connection); } } deque.addLast(message); if (deque.size() == 1) { connection.getOwner() .getWorker() .executeTaskInSequence(() -> updateMessageCallback(message, connection, deque), ThreadsWorker.WorkerType.INBOUND, connection); } } BindingHandler(SxpNode node, BindingDispatcher dispatcher);  BindingHandler(SxpNode node, BindingDispatcher dispatcher, int bufferSize); }
BindingHandler { public <T extends SxpBindingFields> void processUpdate(final List<T> databaseDelete, final List<T> databaseAdd, final SxpConnection connection) { final Deque<DecodedMessage> deque; final DecodedMessage message = new DecodedMessage(databaseDelete, databaseAdd); synchronized (buffer) { if (!buffer.containsKey(connection)) { buffer.put(connection, (deque = new ArrayDeque<>())); } else { deque = buffer.get(connection); } } deque.addLast(message); if (deque.size() == 1) { connection.getOwner() .getWorker() .executeTaskInSequence(() -> updateMessageCallback(message, connection, deque), ThreadsWorker.WorkerType.INBOUND, connection); } } BindingHandler(SxpNode node, BindingDispatcher dispatcher);  BindingHandler(SxpNode node, BindingDispatcher dispatcher, int bufferSize); void setBufferLimit(int limit); static Stream<T> loopDetection(NodeId nodeId, Stream<T> bindings); static List<SxpDatabaseBinding> processMessageAddition(UpdateMessageLegacy message, SxpBindingFilter filter,
            NodeId nodeIdRemote); static List<SxpDatabaseBinding> processMessageAddition(UpdateMessage message, SxpBindingFilter filter); static List<SxpDatabaseBinding> processMessageDeletion(UpdateMessage message); static List<SxpDatabaseBinding> processMessageDeletion(UpdateMessageLegacy message); ListenableFuture<Void> processPurgeAllMessage(final SxpConnection connection); void processUpdate(final List<T> databaseDelete, final List<T> databaseAdd,
            final SxpConnection connection); }
BindingHandler { public <T extends SxpBindingFields> void processUpdate(final List<T> databaseDelete, final List<T> databaseAdd, final SxpConnection connection) { final Deque<DecodedMessage> deque; final DecodedMessage message = new DecodedMessage(databaseDelete, databaseAdd); synchronized (buffer) { if (!buffer.containsKey(connection)) { buffer.put(connection, (deque = new ArrayDeque<>())); } else { deque = buffer.get(connection); } } deque.addLast(message); if (deque.size() == 1) { connection.getOwner() .getWorker() .executeTaskInSequence(() -> updateMessageCallback(message, connection, deque), ThreadsWorker.WorkerType.INBOUND, connection); } } BindingHandler(SxpNode node, BindingDispatcher dispatcher);  BindingHandler(SxpNode node, BindingDispatcher dispatcher, int bufferSize); void setBufferLimit(int limit); static Stream<T> loopDetection(NodeId nodeId, Stream<T> bindings); static List<SxpDatabaseBinding> processMessageAddition(UpdateMessageLegacy message, SxpBindingFilter filter,
            NodeId nodeIdRemote); static List<SxpDatabaseBinding> processMessageAddition(UpdateMessage message, SxpBindingFilter filter); static List<SxpDatabaseBinding> processMessageDeletion(UpdateMessage message); static List<SxpDatabaseBinding> processMessageDeletion(UpdateMessageLegacy message); ListenableFuture<Void> processPurgeAllMessage(final SxpConnection connection); void processUpdate(final List<T> databaseDelete, final List<T> databaseAdd,
            final SxpConnection connection);  }