@Test public void testMute() throws Exception { blockingConnect("0"); blockingConnect("1"); selector.send(createSend("0", "hello")); selector.send(createSend("1", "hi")); selector.mute("1"); while (selector.completedReceives().isEmpty()) selector.poll(5); assertEquals("We should have only one response", 1, selector.completedReceives().size()); assertEquals("The response should not be from the muted node", "0", selector.completedReceives().get(0).source()); selector.unmute("1"); do { selector.poll(5); } while (selector.completedReceives().isEmpty()); assertEquals("We should have only one response", 1, selector.completedReceives().size()); assertEquals("The response should be from the previously muted node", "1", selector.completedReceives().get(0).source()); }
@Override public void mute(String id) { KafkaChannel channel = channelOrFail(id); mute(channel); }
Selector implements Selectable { @Override public void mute(String id) { KafkaChannel channel = channelOrFail(id); mute(channel); } }
Selector implements Selectable { @Override public void mute(String id) { KafkaChannel channel = channelOrFail(id); mute(channel); } Selector(int maxReceiveSize, long connectionMaxIdleMs, Metrics metrics, Time time, String metricGrpPrefix, Map<String, String> metricTags, boolean metricsPerConnection,
        ChannelBuilder channelBuilder);  Selector(long connectionMaxIdleMS, Metrics metrics, Time time, String metricGrpPrefix, ChannelBuilder channelBuilder); }
Selector implements Selectable { @Override public void mute(String id) { KafkaChannel channel = channelOrFail(id); mute(channel); } Selector(int maxReceiveSize, long connectionMaxIdleMs, Metrics metrics, Time time, String metricGrpPrefix, Map<String, String> metricTags, boolean metricsPerConnection,
        ChannelBuilder channelBuilder);  Selector(long connectionMaxIdleMS, Metrics metrics, Time time, String metricGrpPrefix, ChannelBuilder channelBuilder); @Override void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize); void register(String id, SocketChannel socketChannel); @Override void wakeup(); @Override void close(); void send(Send send); @Override void poll(long timeout); static void main(String[] args); @Override List<Send> completedSends(); @Override List<NetworkReceive> completedReceives(); @Override List<String> disconnected(); @Override List<String> connected(); void close(String id); @Override boolean isChannelReady(String id); List<KafkaChannel> channels(); KafkaChannel channel(String id); @Override void mute(String id); @Override void unmute(String id); @Override void muteAll(); @Override void unmuteAll(); }
Selector implements Selectable { @Override public void mute(String id) { KafkaChannel channel = channelOrFail(id); mute(channel); } Selector(int maxReceiveSize, long connectionMaxIdleMs, Metrics metrics, Time time, String metricGrpPrefix, Map<String, String> metricTags, boolean metricsPerConnection,
        ChannelBuilder channelBuilder);  Selector(long connectionMaxIdleMS, Metrics metrics, Time time, String metricGrpPrefix, ChannelBuilder channelBuilder); @Override void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize); void register(String id, SocketChannel socketChannel); @Override void wakeup(); @Override void close(); void send(Send send); @Override void poll(long timeout); static void main(String[] args); @Override List<Send> completedSends(); @Override List<NetworkReceive> completedReceives(); @Override List<String> disconnected(); @Override List<String> connected(); void close(String id); @Override boolean isChannelReady(String id); List<KafkaChannel> channels(); KafkaChannel channel(String id); @Override void mute(String id); @Override void unmute(String id); @Override void muteAll(); @Override void unmuteAll();  }