@Test public void testProcessPurgeAllMessage() throws Exception { handler.processPurgeAllMessage(connection).get(); verify(worker).executeTaskInSequence(any(Callable.class), eq(ThreadsWorker.WorkerType.INBOUND), eq(connection)); }
public ListenableFuture<Void> processPurgeAllMessage(final SxpConnection connection) { synchronized (buffer) { if (buffer.containsKey(connection)) { buffer.get(connection).clear(); } } return Preconditions.checkNotNull(connection).getOwner().getWorker().executeTaskInSequence(() -> { final Map<NodeId, SxpBindingFilter> filterMap = SxpDatabase.getInboundFilters(connection.getOwner(), connection.getDomainName()); final SxpDomain sxpDomain = connection.getOwner().getDomain(connection.getDomainName()); SxpBindingFilter<?, ? extends SxpFilterFields> filter = connection.getFilter(FilterType.Inbound); synchronized (sxpDomain) { List<SxpDatabaseBinding> removed = sxpDomain.getSxpDatabase() .deleteBindings(connection.getId()); List<SxpDatabaseBinding> replace = SxpDatabase .getReplaceForBindings(removed, sxpDomain.getSxpDatabase(), filterMap); sxpDomain.getMasterDatabase().deleteBindings(removed); sxpDomain.getMasterDatabase().addBindings(replace); sxpDomain.pushToSharedSxpDatabases(connection.getId(), filter, removed, replace); } return null; }, ThreadsWorker.WorkerType.INBOUND, connection); }
BindingHandler { public ListenableFuture<Void> processPurgeAllMessage(final SxpConnection connection) { synchronized (buffer) { if (buffer.containsKey(connection)) { buffer.get(connection).clear(); } } return Preconditions.checkNotNull(connection).getOwner().getWorker().executeTaskInSequence(() -> { final Map<NodeId, SxpBindingFilter> filterMap = SxpDatabase.getInboundFilters(connection.getOwner(), connection.getDomainName()); final SxpDomain sxpDomain = connection.getOwner().getDomain(connection.getDomainName()); SxpBindingFilter<?, ? extends SxpFilterFields> filter = connection.getFilter(FilterType.Inbound); synchronized (sxpDomain) { List<SxpDatabaseBinding> removed = sxpDomain.getSxpDatabase() .deleteBindings(connection.getId()); List<SxpDatabaseBinding> replace = SxpDatabase .getReplaceForBindings(removed, sxpDomain.getSxpDatabase(), filterMap); sxpDomain.getMasterDatabase().deleteBindings(removed); sxpDomain.getMasterDatabase().addBindings(replace); sxpDomain.pushToSharedSxpDatabases(connection.getId(), filter, removed, replace); } return null; }, ThreadsWorker.WorkerType.INBOUND, connection); } }
BindingHandler { public ListenableFuture<Void> processPurgeAllMessage(final SxpConnection connection) { synchronized (buffer) { if (buffer.containsKey(connection)) { buffer.get(connection).clear(); } } return Preconditions.checkNotNull(connection).getOwner().getWorker().executeTaskInSequence(() -> { final Map<NodeId, SxpBindingFilter> filterMap = SxpDatabase.getInboundFilters(connection.getOwner(), connection.getDomainName()); final SxpDomain sxpDomain = connection.getOwner().getDomain(connection.getDomainName()); SxpBindingFilter<?, ? extends SxpFilterFields> filter = connection.getFilter(FilterType.Inbound); synchronized (sxpDomain) { List<SxpDatabaseBinding> removed = sxpDomain.getSxpDatabase() .deleteBindings(connection.getId()); List<SxpDatabaseBinding> replace = SxpDatabase .getReplaceForBindings(removed, sxpDomain.getSxpDatabase(), filterMap); sxpDomain.getMasterDatabase().deleteBindings(removed); sxpDomain.getMasterDatabase().addBindings(replace); sxpDomain.pushToSharedSxpDatabases(connection.getId(), filter, removed, replace); } return null; }, ThreadsWorker.WorkerType.INBOUND, connection); } BindingHandler(SxpNode node, BindingDispatcher dispatcher);  BindingHandler(SxpNode node, BindingDispatcher dispatcher, int bufferSize); }
BindingHandler { public ListenableFuture<Void> processPurgeAllMessage(final SxpConnection connection) { synchronized (buffer) { if (buffer.containsKey(connection)) { buffer.get(connection).clear(); } } return Preconditions.checkNotNull(connection).getOwner().getWorker().executeTaskInSequence(() -> { final Map<NodeId, SxpBindingFilter> filterMap = SxpDatabase.getInboundFilters(connection.getOwner(), connection.getDomainName()); final SxpDomain sxpDomain = connection.getOwner().getDomain(connection.getDomainName()); SxpBindingFilter<?, ? extends SxpFilterFields> filter = connection.getFilter(FilterType.Inbound); synchronized (sxpDomain) { List<SxpDatabaseBinding> removed = sxpDomain.getSxpDatabase() .deleteBindings(connection.getId()); List<SxpDatabaseBinding> replace = SxpDatabase .getReplaceForBindings(removed, sxpDomain.getSxpDatabase(), filterMap); sxpDomain.getMasterDatabase().deleteBindings(removed); sxpDomain.getMasterDatabase().addBindings(replace); sxpDomain.pushToSharedSxpDatabases(connection.getId(), filter, removed, replace); } return null; }, 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 ListenableFuture<Void> processPurgeAllMessage(final SxpConnection connection) { synchronized (buffer) { if (buffer.containsKey(connection)) { buffer.get(connection).clear(); } } return Preconditions.checkNotNull(connection).getOwner().getWorker().executeTaskInSequence(() -> { final Map<NodeId, SxpBindingFilter> filterMap = SxpDatabase.getInboundFilters(connection.getOwner(), connection.getDomainName()); final SxpDomain sxpDomain = connection.getOwner().getDomain(connection.getDomainName()); SxpBindingFilter<?, ? extends SxpFilterFields> filter = connection.getFilter(FilterType.Inbound); synchronized (sxpDomain) { List<SxpDatabaseBinding> removed = sxpDomain.getSxpDatabase() .deleteBindings(connection.getId()); List<SxpDatabaseBinding> replace = SxpDatabase .getReplaceForBindings(removed, sxpDomain.getSxpDatabase(), filterMap); sxpDomain.getMasterDatabase().deleteBindings(removed); sxpDomain.getMasterDatabase().addBindings(replace); sxpDomain.pushToSharedSxpDatabases(connection.getId(), filter, removed, replace); } return null; }, 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);  }