@Test public void testCallWithNullMessage() { partitions[0] = (t, u) -> null; new UpdateExportTask(connection, byteBuffs, partitions, atomicInteger).call(); verifyNoMoreInteractions(byteBuffs); }
@Override public Void call() { for (int i = 0; i < partitions.length; i++) { BiFunction<SxpConnection, SxpBindingFilter, ByteBuf> data; synchronized (partitions) { data = partitions[i]; partitions[i] = null; } if (data != null) { ByteBuf message = data.apply(connection, connection.getFilter(FilterType.Outbound)); if (message == null) { LOG.error("{} Generated empty partition.", connection); return null; } synchronized (generatedMessages) { generatedMessages[i] = message; generatedMessages.notifyAll(); } } } try { for (int i = 0; i < generatedMessages.length; i++) { ByteBuf message; do { synchronized (generatedMessages) { if ((message = generatedMessages[i]) == null) { generatedMessages.wait(); } } } while (message == null); if (message.capacity() == 0) { LOG.warn("{} Cannot export empty message aborting export", connection); freeReferences(); return null; } } for (int i = 0; i < generatedMessages.length; i++) { connection.getChannelHandlerContext(SxpConnection.ChannelHandlerContextType.SPEAKER_CNTXT) .write(generatedMessages[i].duplicate().retain()); if (LOG.isTraceEnabled()) { LOG.trace("{} {} UPDATEv{} {}", connection, i, connection.getVersion().getIntValue(), MessageFactory.toString(generatedMessages[i])); } } connection.getChannelHandlerContext(SxpConnection.ChannelHandlerContextType.SPEAKER_CNTXT).flush(); connection.setUpdateOrKeepaliveMessageTimestamp(); } catch (ChannelHandlerContextNotFoundException | ChannelHandlerContextDiscrepancyException e) { LOG.warn("{} Cannot find context aborting bindings export.", connection); } catch (InterruptedException e) { LOG.warn("{} Bindings export canceled.", connection, e); } freeReferences(); return null; }
UpdateExportTask implements Callable<Void> { @Override public Void call() { for (int i = 0; i < partitions.length; i++) { BiFunction<SxpConnection, SxpBindingFilter, ByteBuf> data; synchronized (partitions) { data = partitions[i]; partitions[i] = null; } if (data != null) { ByteBuf message = data.apply(connection, connection.getFilter(FilterType.Outbound)); if (message == null) { LOG.error("{} Generated empty partition.", connection); return null; } synchronized (generatedMessages) { generatedMessages[i] = message; generatedMessages.notifyAll(); } } } try { for (int i = 0; i < generatedMessages.length; i++) { ByteBuf message; do { synchronized (generatedMessages) { if ((message = generatedMessages[i]) == null) { generatedMessages.wait(); } } } while (message == null); if (message.capacity() == 0) { LOG.warn("{} Cannot export empty message aborting export", connection); freeReferences(); return null; } } for (int i = 0; i < generatedMessages.length; i++) { connection.getChannelHandlerContext(SxpConnection.ChannelHandlerContextType.SPEAKER_CNTXT) .write(generatedMessages[i].duplicate().retain()); if (LOG.isTraceEnabled()) { LOG.trace("{} {} UPDATEv{} {}", connection, i, connection.getVersion().getIntValue(), MessageFactory.toString(generatedMessages[i])); } } connection.getChannelHandlerContext(SxpConnection.ChannelHandlerContextType.SPEAKER_CNTXT).flush(); connection.setUpdateOrKeepaliveMessageTimestamp(); } catch (ChannelHandlerContextNotFoundException | ChannelHandlerContextDiscrepancyException e) { LOG.warn("{} Cannot find context aborting bindings export.", connection); } catch (InterruptedException e) { LOG.warn("{} Bindings export canceled.", connection, e); } freeReferences(); return null; } }
UpdateExportTask implements Callable<Void> { @Override public Void call() { for (int i = 0; i < partitions.length; i++) { BiFunction<SxpConnection, SxpBindingFilter, ByteBuf> data; synchronized (partitions) { data = partitions[i]; partitions[i] = null; } if (data != null) { ByteBuf message = data.apply(connection, connection.getFilter(FilterType.Outbound)); if (message == null) { LOG.error("{} Generated empty partition.", connection); return null; } synchronized (generatedMessages) { generatedMessages[i] = message; generatedMessages.notifyAll(); } } } try { for (int i = 0; i < generatedMessages.length; i++) { ByteBuf message; do { synchronized (generatedMessages) { if ((message = generatedMessages[i]) == null) { generatedMessages.wait(); } } } while (message == null); if (message.capacity() == 0) { LOG.warn("{} Cannot export empty message aborting export", connection); freeReferences(); return null; } } for (int i = 0; i < generatedMessages.length; i++) { connection.getChannelHandlerContext(SxpConnection.ChannelHandlerContextType.SPEAKER_CNTXT) .write(generatedMessages[i].duplicate().retain()); if (LOG.isTraceEnabled()) { LOG.trace("{} {} UPDATEv{} {}", connection, i, connection.getVersion().getIntValue(), MessageFactory.toString(generatedMessages[i])); } } connection.getChannelHandlerContext(SxpConnection.ChannelHandlerContextType.SPEAKER_CNTXT).flush(); connection.setUpdateOrKeepaliveMessageTimestamp(); } catch (ChannelHandlerContextNotFoundException | ChannelHandlerContextDiscrepancyException e) { LOG.warn("{} Cannot find context aborting bindings export.", connection); } catch (InterruptedException e) { LOG.warn("{} Bindings export canceled.", connection, e); } freeReferences(); return null; } UpdateExportTask(SxpConnection connection, ByteBuf[] generatedMessages,
            BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>[] partitions, AtomicInteger messagesReleaseCounter); }
UpdateExportTask implements Callable<Void> { @Override public Void call() { for (int i = 0; i < partitions.length; i++) { BiFunction<SxpConnection, SxpBindingFilter, ByteBuf> data; synchronized (partitions) { data = partitions[i]; partitions[i] = null; } if (data != null) { ByteBuf message = data.apply(connection, connection.getFilter(FilterType.Outbound)); if (message == null) { LOG.error("{} Generated empty partition.", connection); return null; } synchronized (generatedMessages) { generatedMessages[i] = message; generatedMessages.notifyAll(); } } } try { for (int i = 0; i < generatedMessages.length; i++) { ByteBuf message; do { synchronized (generatedMessages) { if ((message = generatedMessages[i]) == null) { generatedMessages.wait(); } } } while (message == null); if (message.capacity() == 0) { LOG.warn("{} Cannot export empty message aborting export", connection); freeReferences(); return null; } } for (int i = 0; i < generatedMessages.length; i++) { connection.getChannelHandlerContext(SxpConnection.ChannelHandlerContextType.SPEAKER_CNTXT) .write(generatedMessages[i].duplicate().retain()); if (LOG.isTraceEnabled()) { LOG.trace("{} {} UPDATEv{} {}", connection, i, connection.getVersion().getIntValue(), MessageFactory.toString(generatedMessages[i])); } } connection.getChannelHandlerContext(SxpConnection.ChannelHandlerContextType.SPEAKER_CNTXT).flush(); connection.setUpdateOrKeepaliveMessageTimestamp(); } catch (ChannelHandlerContextNotFoundException | ChannelHandlerContextDiscrepancyException e) { LOG.warn("{} Cannot find context aborting bindings export.", connection); } catch (InterruptedException e) { LOG.warn("{} Bindings export canceled.", connection, e); } freeReferences(); return null; } UpdateExportTask(SxpConnection connection, ByteBuf[] generatedMessages,
            BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>[] partitions, AtomicInteger messagesReleaseCounter); SxpConnection getConnection(); @Override Void call(); void freeReferences(); }
UpdateExportTask implements Callable<Void> { @Override public Void call() { for (int i = 0; i < partitions.length; i++) { BiFunction<SxpConnection, SxpBindingFilter, ByteBuf> data; synchronized (partitions) { data = partitions[i]; partitions[i] = null; } if (data != null) { ByteBuf message = data.apply(connection, connection.getFilter(FilterType.Outbound)); if (message == null) { LOG.error("{} Generated empty partition.", connection); return null; } synchronized (generatedMessages) { generatedMessages[i] = message; generatedMessages.notifyAll(); } } } try { for (int i = 0; i < generatedMessages.length; i++) { ByteBuf message; do { synchronized (generatedMessages) { if ((message = generatedMessages[i]) == null) { generatedMessages.wait(); } } } while (message == null); if (message.capacity() == 0) { LOG.warn("{} Cannot export empty message aborting export", connection); freeReferences(); return null; } } for (int i = 0; i < generatedMessages.length; i++) { connection.getChannelHandlerContext(SxpConnection.ChannelHandlerContextType.SPEAKER_CNTXT) .write(generatedMessages[i].duplicate().retain()); if (LOG.isTraceEnabled()) { LOG.trace("{} {} UPDATEv{} {}", connection, i, connection.getVersion().getIntValue(), MessageFactory.toString(generatedMessages[i])); } } connection.getChannelHandlerContext(SxpConnection.ChannelHandlerContextType.SPEAKER_CNTXT).flush(); connection.setUpdateOrKeepaliveMessageTimestamp(); } catch (ChannelHandlerContextNotFoundException | ChannelHandlerContextDiscrepancyException e) { LOG.warn("{} Cannot find context aborting bindings export.", connection); } catch (InterruptedException e) { LOG.warn("{} Bindings export canceled.", connection, e); } freeReferences(); return null; } UpdateExportTask(SxpConnection connection, ByteBuf[] generatedMessages,
            BiFunction<SxpConnection, SxpBindingFilter, ByteBuf>[] partitions, AtomicInteger messagesReleaseCounter); SxpConnection getConnection(); @Override Void call(); void freeReferences();  }