@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, true); mute(channel); }
Selector implements Selectable, AutoCloseable { @Override public void mute(String id) { KafkaChannel channel = channelOrFail(id, true); mute(channel); } }
Selector implements Selectable, AutoCloseable { @Override public void mute(String id) { KafkaChannel channel = channelOrFail(id, true); mute(channel); } Selector(int maxReceiveSize,
                    long connectionMaxIdleMs,
                    Metrics metrics,
                    Time time,
                    String metricGrpPrefix, // group的前缀
                    Map<String, String> metricTags, // 创建MetricName时候使用的tags集合
                    boolean metricsPerConnection,
                    boolean recordTimePerConnection,
                    ChannelBuilder channelBuilder);  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, AutoCloseable { @Override public void mute(String id) { KafkaChannel channel = channelOrFail(id, true); mute(channel); } Selector(int maxReceiveSize,
                    long connectionMaxIdleMs,
                    Metrics metrics,
                    Time time,
                    String metricGrpPrefix, // group的前缀
                    Map<String, String> metricTags, // 创建MetricName时候使用的tags集合
                    boolean metricsPerConnection,
                    boolean recordTimePerConnection,
                    ChannelBuilder channelBuilder);  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 // 创建 KafkaChannel添加到channels保存 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 // 轮训的时候根据选在键读写，分别调用kafka通道的read和write void poll(long timeout); @Override List<Send> completedSends(); @Override List<NetworkReceive> completedReceives(); @Override Map<String, ChannelState> disconnected(); @Override List<String> connected(); @Override void mute(String id); @Override void unmute(String id); @Override void muteAll(); @Override void unmuteAll(); void close(String id); @Override boolean isChannelReady(String id); List<KafkaChannel> channels(); KafkaChannel channel(String id); KafkaChannel closingChannel(String id); Set<SelectionKey> keys(); }
Selector implements Selectable, AutoCloseable { @Override public void mute(String id) { KafkaChannel channel = channelOrFail(id, true); mute(channel); } Selector(int maxReceiveSize,
                    long connectionMaxIdleMs,
                    Metrics metrics,
                    Time time,
                    String metricGrpPrefix, // group的前缀
                    Map<String, String> metricTags, // 创建MetricName时候使用的tags集合
                    boolean metricsPerConnection,
                    boolean recordTimePerConnection,
                    ChannelBuilder channelBuilder);  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 // 创建 KafkaChannel添加到channels保存 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 // 轮训的时候根据选在键读写，分别调用kafka通道的read和write void poll(long timeout); @Override List<Send> completedSends(); @Override List<NetworkReceive> completedReceives(); @Override Map<String, ChannelState> disconnected(); @Override List<String> connected(); @Override void mute(String id); @Override void unmute(String id); @Override void muteAll(); @Override void unmuteAll(); void close(String id); @Override boolean isChannelReady(String id); List<KafkaChannel> channels(); KafkaChannel channel(String id); KafkaChannel closingChannel(String id); Set<SelectionKey> keys(); static final long NO_IDLE_TIMEOUT_MS; }