@Test public void testPropagateUpdateWithEmptyInputs() { dispatcher.propagateUpdate(null, null, null); dispatcher.propagateUpdate(Collections.EMPTY_LIST, null, null); dispatcher.propagateUpdate(Collections.EMPTY_LIST, Collections.EMPTY_LIST, null); dispatcher.propagateUpdate(Collections.EMPTY_LIST, Collections.EMPTY_LIST, Collections.EMPTY_LIST); verify(worker, never()).executeTaskInSequence(any(Callable.class), eq(ThreadsWorker.WorkerType.OUTBOUND), any(SxpConnection.class)); }
public <T extends SxpBindingFields> void propagateUpdate(List<T> deleteBindings, List<T> addBindings, List<SxpConnection> connections) { if ((deleteBindings == null || deleteBindings.isEmpty()) && (addBindings == null || addBindings.isEmpty()) || ( connections == null || connections.isEmpty())) { return; } Map<ExportKey, BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>[]> dataPool = new HashMap<>(4); Map<ExportKey, ByteBuf[]> messagesPool = new HashMap<>(4); Map<ExportKey, AtomicInteger> releaseCounterPool = new HashMap<>(4); List<UpdateExportTask> exportTasks = new ArrayList<>(); for (SxpConnection connection : connections) { if (!connection.isStateOn() || !connection.isModeSpeaker()) { continue; } ExportKey key = new ExportKey(connection); if (dataPool.get(key) == null) { List<BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>> partitions = partitionBindings(connection, deleteBindings, addBindings); dataPool.put(key, partitions.toArray(new BiFunction[partitions.size()])); messagesPool.put(key, new ByteBuf[partitions.size()]); releaseCounterPool.put(key, new AtomicInteger(0)); } AtomicInteger releaseCounter = releaseCounterPool.get(key); releaseCounter.incrementAndGet(); exportTasks.add(new UpdateExportTask(connection, messagesPool.get(key), dataPool.get(key), releaseCounter)); } exportTasks.forEach(e -> worker.executeTaskInSequence(e, ThreadsWorker.WorkerType.OUTBOUND, e.getConnection())); }
BindingDispatcher { public <T extends SxpBindingFields> void propagateUpdate(List<T> deleteBindings, List<T> addBindings, List<SxpConnection> connections) { if ((deleteBindings == null || deleteBindings.isEmpty()) && (addBindings == null || addBindings.isEmpty()) || ( connections == null || connections.isEmpty())) { return; } Map<ExportKey, BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>[]> dataPool = new HashMap<>(4); Map<ExportKey, ByteBuf[]> messagesPool = new HashMap<>(4); Map<ExportKey, AtomicInteger> releaseCounterPool = new HashMap<>(4); List<UpdateExportTask> exportTasks = new ArrayList<>(); for (SxpConnection connection : connections) { if (!connection.isStateOn() || !connection.isModeSpeaker()) { continue; } ExportKey key = new ExportKey(connection); if (dataPool.get(key) == null) { List<BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>> partitions = partitionBindings(connection, deleteBindings, addBindings); dataPool.put(key, partitions.toArray(new BiFunction[partitions.size()])); messagesPool.put(key, new ByteBuf[partitions.size()]); releaseCounterPool.put(key, new AtomicInteger(0)); } AtomicInteger releaseCounter = releaseCounterPool.get(key); releaseCounter.incrementAndGet(); exportTasks.add(new UpdateExportTask(connection, messagesPool.get(key), dataPool.get(key), releaseCounter)); } exportTasks.forEach(e -> worker.executeTaskInSequence(e, ThreadsWorker.WorkerType.OUTBOUND, e.getConnection())); } }
BindingDispatcher { public <T extends SxpBindingFields> void propagateUpdate(List<T> deleteBindings, List<T> addBindings, List<SxpConnection> connections) { if ((deleteBindings == null || deleteBindings.isEmpty()) && (addBindings == null || addBindings.isEmpty()) || ( connections == null || connections.isEmpty())) { return; } Map<ExportKey, BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>[]> dataPool = new HashMap<>(4); Map<ExportKey, ByteBuf[]> messagesPool = new HashMap<>(4); Map<ExportKey, AtomicInteger> releaseCounterPool = new HashMap<>(4); List<UpdateExportTask> exportTasks = new ArrayList<>(); for (SxpConnection connection : connections) { if (!connection.isStateOn() || !connection.isModeSpeaker()) { continue; } ExportKey key = new ExportKey(connection); if (dataPool.get(key) == null) { List<BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>> partitions = partitionBindings(connection, deleteBindings, addBindings); dataPool.put(key, partitions.toArray(new BiFunction[partitions.size()])); messagesPool.put(key, new ByteBuf[partitions.size()]); releaseCounterPool.put(key, new AtomicInteger(0)); } AtomicInteger releaseCounter = releaseCounterPool.get(key); releaseCounter.incrementAndGet(); exportTasks.add(new UpdateExportTask(connection, messagesPool.get(key), dataPool.get(key), releaseCounter)); } exportTasks.forEach(e -> worker.executeTaskInSequence(e, ThreadsWorker.WorkerType.OUTBOUND, e.getConnection())); } BindingDispatcher(SxpNode owner); }
BindingDispatcher { public <T extends SxpBindingFields> void propagateUpdate(List<T> deleteBindings, List<T> addBindings, List<SxpConnection> connections) { if ((deleteBindings == null || deleteBindings.isEmpty()) && (addBindings == null || addBindings.isEmpty()) || ( connections == null || connections.isEmpty())) { return; } Map<ExportKey, BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>[]> dataPool = new HashMap<>(4); Map<ExportKey, ByteBuf[]> messagesPool = new HashMap<>(4); Map<ExportKey, AtomicInteger> releaseCounterPool = new HashMap<>(4); List<UpdateExportTask> exportTasks = new ArrayList<>(); for (SxpConnection connection : connections) { if (!connection.isStateOn() || !connection.isModeSpeaker()) { continue; } ExportKey key = new ExportKey(connection); if (dataPool.get(key) == null) { List<BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>> partitions = partitionBindings(connection, deleteBindings, addBindings); dataPool.put(key, partitions.toArray(new BiFunction[partitions.size()])); messagesPool.put(key, new ByteBuf[partitions.size()]); releaseCounterPool.put(key, new AtomicInteger(0)); } AtomicInteger releaseCounter = releaseCounterPool.get(key); releaseCounter.incrementAndGet(); exportTasks.add(new UpdateExportTask(connection, messagesPool.get(key), dataPool.get(key), releaseCounter)); } exportTasks.forEach(e -> worker.executeTaskInSequence(e, ThreadsWorker.WorkerType.OUTBOUND, e.getConnection())); } BindingDispatcher(SxpNode owner); void setPartitionSize(int partitionSize); void propagateUpdate(List<T> deleteBindings, List<T> addBindings,
                                                             List<SxpConnection> connections); static ListenableFuture<Boolean> sendPurgeAllMessage(final SxpConnection connection); static boolean sendPurgeAllMessageSync(final SxpConnection connection); SxpNode getOwner(); }
BindingDispatcher { public <T extends SxpBindingFields> void propagateUpdate(List<T> deleteBindings, List<T> addBindings, List<SxpConnection> connections) { if ((deleteBindings == null || deleteBindings.isEmpty()) && (addBindings == null || addBindings.isEmpty()) || ( connections == null || connections.isEmpty())) { return; } Map<ExportKey, BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>[]> dataPool = new HashMap<>(4); Map<ExportKey, ByteBuf[]> messagesPool = new HashMap<>(4); Map<ExportKey, AtomicInteger> releaseCounterPool = new HashMap<>(4); List<UpdateExportTask> exportTasks = new ArrayList<>(); for (SxpConnection connection : connections) { if (!connection.isStateOn() || !connection.isModeSpeaker()) { continue; } ExportKey key = new ExportKey(connection); if (dataPool.get(key) == null) { List<BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>> partitions = partitionBindings(connection, deleteBindings, addBindings); dataPool.put(key, partitions.toArray(new BiFunction[partitions.size()])); messagesPool.put(key, new ByteBuf[partitions.size()]); releaseCounterPool.put(key, new AtomicInteger(0)); } AtomicInteger releaseCounter = releaseCounterPool.get(key); releaseCounter.incrementAndGet(); exportTasks.add(new UpdateExportTask(connection, messagesPool.get(key), dataPool.get(key), releaseCounter)); } exportTasks.forEach(e -> worker.executeTaskInSequence(e, ThreadsWorker.WorkerType.OUTBOUND, e.getConnection())); } BindingDispatcher(SxpNode owner); void setPartitionSize(int partitionSize); void propagateUpdate(List<T> deleteBindings, List<T> addBindings,
                                                             List<SxpConnection> connections); static ListenableFuture<Boolean> sendPurgeAllMessage(final SxpConnection connection); static boolean sendPurgeAllMessageSync(final SxpConnection connection); SxpNode getOwner();  }