fix: make sure FIFO order for write() when notifyChannelActive(), also make sure channel access thread-safe and avoid potential NPE

okg-cxf · okg-cxf · commit d23c68f6de67 · 2024-01-18T14:17:44.000+08:00
diff --git a/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java b/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java
@@ -63,11 +63,11 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
 
     private static final AtomicLong ENDPOINT_COUNTER = new AtomicLong();
 
-    private static final AtomicIntegerFieldUpdater<DefaultEndpoint> QUEUE_SIZE = AtomicIntegerFieldUpdater
-            .newUpdater(DefaultEndpoint.class, "queueSize");
+    private static final AtomicIntegerFieldUpdater<DefaultEndpoint> QUEUE_SIZE = AtomicIntegerFieldUpdater.newUpdater(
+            DefaultEndpoint.class, "queueSize");
 
-    private static final AtomicIntegerFieldUpdater<DefaultEndpoint> STATUS = AtomicIntegerFieldUpdater
-            .newUpdater(DefaultEndpoint.class, "status");
+    private static final AtomicIntegerFieldUpdater<DefaultEndpoint> STATUS = AtomicIntegerFieldUpdater.newUpdater(
+            DefaultEndpoint.class, "status");
 
     private static final int ST_OPEN = 0;
 
@@ -191,9 +191,9 @@ public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {
             }
 
             if (autoFlushCommands) {
-
-                if (isConnected()) {
-                    writeToChannelAndFlush(command);
+                Channel channel = this.channel;
+                if (isConnected(channel)) {
+                    writeToChannelAndFlush(channel, command);
                 } else {
                     writeToDisconnectedBuffer(command);
                 }
@@ -232,9 +232,9 @@ public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {
             }
 
             if (autoFlushCommands) {
-
-                if (isConnected()) {
-                    writeToChannelAndFlush(commands);
+                Channel channel = this.channel;
+                if (isConnected(channel)) {
+                    writeToChannelAndFlush(channel, commands);
                 } else {
                     writeToDisconnectedBuffer(commands);
                 }
@@ -284,10 +284,9 @@ private RedisException validateWrite(int commands) {
             return new RedisException("Connection is closed");
         }
 
+        final boolean connected = isConnected(this.channel);
         if (usesBoundedQueues()) {
 
-            boolean connected = isConnected();
-
             if (QUEUE_SIZE.get(this) + commands > clientOptions.getRequestQueueSize()) {
                 return new RedisException("Request queue size exceeded: " + clientOptions.getRequestQueueSize()
                         + ". Commands are not accepted until the queue size drops.");
@@ -304,7 +303,7 @@ private RedisException validateWrite(int commands) {
             }
         }
 
-        if (!isConnected() && rejectCommandsWhileDisconnected) {
+        if (!connected && rejectCommandsWhileDisconnected) {
             return new RedisException("Currently not connected. Commands are rejected.");
         }
 
@@ -366,11 +365,11 @@ private void writeToDisconnectedBuffer(RedisCommand<?, ?, ?> command) {
         commandBuffer.add(command);
     }
 
-    private void writeToChannelAndFlush(RedisCommand<?, ?, ?> command) {
+    private void writeToChannelAndFlush(Channel channel, RedisCommand<?, ?, ?> command) {
 
         QUEUE_SIZE.incrementAndGet(this);
 
-        ChannelFuture channelFuture = channelWriteAndFlush(command);
+        ChannelFuture channelFuture = channelWriteAndFlush(channel, command);
 
         if (reliability == Reliability.AT_MOST_ONCE) {
             // cancel on exceptions and remove from queue, because there is no housekeeping
@@ -383,30 +382,30 @@ private void writeToChannelAndFlush(RedisCommand<?, ?, ?> command) {
         }
     }
 
-    private void writeToChannelAndFlush(Collection<? extends RedisCommand<?, ?, ?>> commands) {
+    private void writeToChannelAndFlush(Channel channel, Collection<? extends RedisCommand<?, ?, ?>> commands) {
 
         QUEUE_SIZE.addAndGet(this, commands.size());
 
         if (reliability == Reliability.AT_MOST_ONCE) {
 
             // cancel on exceptions and remove from queue, because there is no housekeeping
             for (RedisCommand<?, ?, ?> command : commands) {
-                channelWrite(command).addListener(AtMostOnceWriteListener.newInstance(this, command));
+                channelWrite(channel, command).addListener(AtMostOnceWriteListener.newInstance(this, command));
             }
         }
 
         if (reliability == Reliability.AT_LEAST_ONCE) {
 
             // commands are ok to stay within the queue, reconnect will retrigger them
             for (RedisCommand<?, ?, ?> command : commands) {
-                channelWrite(command).addListener(RetryListener.newInstance(this, command));
+                channelWrite(channel, command).addListener(RetryListener.newInstance(this, command));
             }
         }
 
-        channelFlush();
+        channelFlush(channel);
     }
 
-    private void channelFlush() {
+    private void channelFlush(Channel channel) {
 
         if (debugEnabled) {
             logger.debug("{} write() channelFlush", logPrefix());
@@ -415,7 +414,7 @@ private void channelFlush() {
         channel.flush();
     }
 
-    private ChannelFuture channelWrite(RedisCommand<?, ?, ?> command) {
+    private ChannelFuture channelWrite(Channel channel, RedisCommand<?, ?, ?> command) {
 
         if (debugEnabled) {
             logger.debug("{} write() channelWrite command {}", logPrefix(), command);
@@ -424,7 +423,7 @@ private ChannelFuture channelWrite(RedisCommand<?, ?, ?> command) {
         return channel.write(command);
     }
 
-    private ChannelFuture channelWriteAndFlush(RedisCommand<?, ?, ?> command) {
+    private ChannelFuture channelWriteAndFlush(Channel channel, RedisCommand<?, ?, ?> command) {
 
         if (debugEnabled) {
             logger.debug("{} write() writeAndFlush command {}", logPrefix(), command);
@@ -437,7 +436,6 @@ private ChannelFuture channelWriteAndFlush(RedisCommand<?, ?, ?> command) {
     public void notifyChannelActive(Channel channel) {
 
         this.logPrefix = null;
-        this.channel = channel;
         this.connectionError = null;
 
         if (isClosed()) {
@@ -452,6 +450,7 @@ public void notifyChannelActive(Channel channel) {
         }
 
         sharedLock.doExclusive(() -> {
+            this.channel = channel;
 
             try {
                 // Move queued commands to buffer before issuing any commands because of connection activation.
@@ -474,7 +473,7 @@ public void notifyChannelActive(Channel channel) {
                     inActivation = false;
                 }
 
-                flushCommands(disconnectedBuffer);
+                flushCommands(channel, disconnectedBuffer);
             } catch (Exception e) {
 
                 if (debugEnabled) {
@@ -527,7 +526,7 @@ public void notifyException(Throwable t) {
             doExclusive(this::drainCommands).forEach(cmd -> cmd.completeExceptionally(t));
         }
 
-        if (!isConnected()) {
+        if (!isConnected(this.channel)) {
             connectionError = t;
         }
     }
@@ -540,16 +539,16 @@ public void registerConnectionWatchdog(ConnectionWatchdog connectionWatchdog) {
     @Override
     @SuppressWarnings({ "rawtypes", "unchecked" })
     public void flushCommands() {
-        flushCommands(commandBuffer);
+        flushCommands(this.channel, commandBuffer);
     }
 
-    private void flushCommands(Queue<RedisCommand<?, ?, ?>> queue) {
+    private void flushCommands(Channel channel, Queue<RedisCommand<?, ?, ?>> queue) {
 
         if (debugEnabled) {
             logger.debug("{} flushCommands()", logPrefix());
         }
 
-        if (isConnected()) {
+        if (isConnected(channel)) {
 
             List<RedisCommand<?, ?, ?>> commands = sharedLock.doExclusive(() -> {
 
@@ -565,7 +564,7 @@ private void flushCommands(Queue<RedisCommand<?, ?, ?>> queue) {
             }
 
             if (!commands.isEmpty()) {
-                writeToChannelAndFlush(commands);
+                writeToChannelAndFlush(channel, commands);
             }
         }
     }
@@ -628,10 +627,10 @@ public void disconnect() {
 
     private Channel getOpenChannel() {
 
-        Channel currentChannel = this.channel;
+        Channel channel = this.channel;
 
-        if (currentChannel != null) {
-            return currentChannel;
+        if (channel != null /* && channel.isOpen() is this deliberately omitted? */) {
+            return channel;
         }
 
         return null;
@@ -648,6 +647,7 @@ public void reset() {
             logger.debug("{} reset()", logPrefix());
         }
 
+        Channel channel = this.channel;
         if (channel != null) {
             channel.pipeline().fireUserEventTriggered(new ConnectionEvents.Reset());
         }
@@ -720,9 +720,7 @@ public void notifyDrainQueuedCommands(HasQueuedCommands queuedCommands) {
                 }
             }
 
-            if (isConnected()) {
-                flushCommands(disconnectedBuffer);
-            }
+            flushCommands(this.channel, disconnectedBuffer);
         });
     }
 
@@ -787,9 +785,7 @@ private void cancelCommands(String message, Iterable<? extends RedisCommand<?, ?
         }
     }
 
-    private boolean isConnected() {
-
-        Channel channel = this.channel;
+    private boolean isConnected(Channel channel) {
         return channel != null && channel.isActive();
     }
 
diff --git a/src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java b/src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java
@@ -136,6 +136,34 @@ void before() {
         sut.setConnectionFacade(connectionFacade);
     }
 
+    @Test
+    void writeShouldGuaranteeFIFOOrder() {
+        sut.write(Collections.singletonList(new Command<>(CommandType.SELECT, new StatusOutput<>(StringCodec.UTF8))));
+
+        sut.registerConnectionWatchdog(connectionWatchdog);
+        doAnswer(i -> sut.write(new Command<>(CommandType.AUTH, new StatusOutput<>(StringCodec.UTF8)))).when(connectionWatchdog)
+                .arm();
+        when(channel.isActive()).thenReturn(true);
+
+        sut.notifyChannelActive(channel);
+
+        DefaultChannelPromise promise = new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE);
+
+        when(channel.writeAndFlush(any())).thenAnswer(invocation -> {
+            if (invocation.getArguments()[0] instanceof RedisCommand) {
+                queue.add((RedisCommand) invocation.getArguments()[0]);
+            }
+
+            if (invocation.getArguments()[0] instanceof Collection) {
+                queue.addAll((Collection) invocation.getArguments()[0]);
+            }
+            return promise;
+        });
+
+        assertThat(queue).hasSize(2).first().hasFieldOrPropertyWithValue("type", CommandType.SELECT);
+        assertThat(queue).hasSize(2).last().hasFieldOrPropertyWithValue("type", CommandType.AUTH);
+    }
+
     @Test
     void writeConnectedShouldWriteCommandToChannel() {
 
@@ -396,11 +424,9 @@ void shouldNotReplayActivationCommands() {
 
         when(channel.isActive()).thenReturn(true);
         ConnectionTestUtil.getDisconnectedBuffer(sut)
-                .add(new ActivationCommand<>(
-                new Command<>(CommandType.SELECT, new StatusOutput<>(StringCodec.UTF8))));
+                .add(new ActivationCommand<>(new Command<>(CommandType.SELECT, new StatusOutput<>(StringCodec.UTF8))));
         ConnectionTestUtil.getDisconnectedBuffer(sut).add(new LatencyMeteredCommand<>(
-                new ActivationCommand<>(
-                new Command<>(CommandType.SUBSCRIBE, new StatusOutput<>(StringCodec.UTF8)))));
+                new ActivationCommand<>(new Command<>(CommandType.SUBSCRIBE, new StatusOutput<>(StringCodec.UTF8)))));
 
         doAnswer(i -> {
 

Original file line number	Diff line number	Diff line change
`@@ -63,11 +63,11 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle`
`63`	`63`
`64`	`64`	`private static final AtomicLong ENDPOINT_COUNTER = new AtomicLong();`
`65`	`65`
`66`		`- private static final AtomicIntegerFieldUpdater<DefaultEndpoint> QUEUE_SIZE = AtomicIntegerFieldUpdater`
`67`		`- .newUpdater(DefaultEndpoint.class, "queueSize");`
	`66`	`+ private static final AtomicIntegerFieldUpdater<DefaultEndpoint> QUEUE_SIZE = AtomicIntegerFieldUpdater.newUpdater(`
	`67`	`+ DefaultEndpoint.class, "queueSize");`
`68`	`68`
`69`		`- private static final AtomicIntegerFieldUpdater<DefaultEndpoint> STATUS = AtomicIntegerFieldUpdater`
`70`		`- .newUpdater(DefaultEndpoint.class, "status");`
	`69`	`+ private static final AtomicIntegerFieldUpdater<DefaultEndpoint> STATUS = AtomicIntegerFieldUpdater.newUpdater(`
	`70`	`+ DefaultEndpoint.class, "status");`
`71`	`71`
`72`	`72`	`private static final int ST_OPEN = 0;`
`73`	`73`
`@@ -191,9 +191,9 @@ public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {`
`191`	`191`	`}`
`192`	`192`
`193`	`193`	`if (autoFlushCommands) {`
`194`		`-`
`195`		`- if (isConnected()) {`
`196`		`- writeToChannelAndFlush(command);`
	`194`	`+ Channel channel = this.channel;`
	`195`	`+ if (isConnected(channel)) {`
	`196`	`+ writeToChannelAndFlush(channel, command);`
`197`	`197`	`} else {`
`198`	`198`	`writeToDisconnectedBuffer(command);`
`199`	`199`	`}`
`@@ -232,9 +232,9 @@ public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {`
`232`	`232`	`}`
`233`	`233`
`234`	`234`	`if (autoFlushCommands) {`
`235`		`-`
`236`		`- if (isConnected()) {`
`237`		`- writeToChannelAndFlush(commands);`
	`235`	`+ Channel channel = this.channel;`
	`236`	`+ if (isConnected(channel)) {`
	`237`	`+ writeToChannelAndFlush(channel, commands);`
`238`	`238`	`} else {`
`239`	`239`	`writeToDisconnectedBuffer(commands);`
`240`	`240`	`}`
`@@ -284,10 +284,9 @@ private RedisException validateWrite(int commands) {`
`284`	`284`	`return new RedisException("Connection is closed");`
`285`	`285`	`}`
`286`	`286`
	`287`	`+ final boolean connected = isConnected(this.channel);`
`287`	`288`	`if (usesBoundedQueues()) {`
`288`	`289`
`289`		`- boolean connected = isConnected();`
`290`		`-`
`291`	`290`	`if (QUEUE_SIZE.get(this) + commands > clientOptions.getRequestQueueSize()) {`
`292`	`291`	`return new RedisException("Request queue size exceeded: " + clientOptions.getRequestQueueSize()`
`293`	`292`	`+ ". Commands are not accepted until the queue size drops.");`
`@@ -304,7 +303,7 @@ private RedisException validateWrite(int commands) {`
`304`	`303`	`}`
`305`	`304`	`}`
`306`	`305`
`307`		`- if (!isConnected() && rejectCommandsWhileDisconnected) {`
	`306`	`+ if (!connected && rejectCommandsWhileDisconnected) {`
`308`	`307`	`return new RedisException("Currently not connected. Commands are rejected.");`
`309`	`308`	`}`
`310`	`309`
`@@ -366,11 +365,11 @@ private void writeToDisconnectedBuffer(RedisCommand<?, ?, ?> command) {`
`366`	`365`	`commandBuffer.add(command);`
`367`	`366`	`}`
`368`	`367`
`369`		`- private void writeToChannelAndFlush(RedisCommand<?, ?, ?> command) {`
	`368`	`+ private void writeToChannelAndFlush(Channel channel, RedisCommand<?, ?, ?> command) {`
`370`	`369`
`371`	`370`	`QUEUE_SIZE.incrementAndGet(this);`
`372`	`371`
`373`		`- ChannelFuture channelFuture = channelWriteAndFlush(command);`
	`372`	`+ ChannelFuture channelFuture = channelWriteAndFlush(channel, command);`
`374`	`373`
`375`	`374`	`if (reliability == Reliability.AT_MOST_ONCE) {`
`376`	`375`	`// cancel on exceptions and remove from queue, because there is no housekeeping`
`@@ -383,30 +382,30 @@ private void writeToChannelAndFlush(RedisCommand<?, ?, ?> command) {`
`383`	`382`	`}`
`384`	`383`	`}`
`385`	`384`
`386`		`- private void writeToChannelAndFlush(Collection<? extends RedisCommand<?, ?, ?>> commands) {`
	`385`	`+ private void writeToChannelAndFlush(Channel channel, Collection<? extends RedisCommand<?, ?, ?>> commands) {`
`387`	`386`
`388`	`387`	`QUEUE_SIZE.addAndGet(this, commands.size());`
`389`	`388`
`390`	`389`	`if (reliability == Reliability.AT_MOST_ONCE) {`
`391`	`390`
`392`	`391`	`// cancel on exceptions and remove from queue, because there is no housekeeping`
`393`	`392`	`for (RedisCommand<?, ?, ?> command : commands) {`
`394`		`- channelWrite(command).addListener(AtMostOnceWriteListener.newInstance(this, command));`
	`393`	`+ channelWrite(channel, command).addListener(AtMostOnceWriteListener.newInstance(this, command));`
`395`	`394`	`}`
`396`	`395`	`}`
`397`	`396`
`398`	`397`	`if (reliability == Reliability.AT_LEAST_ONCE) {`
`399`	`398`
`400`	`399`	`// commands are ok to stay within the queue, reconnect will retrigger them`
`401`	`400`	`for (RedisCommand<?, ?, ?> command : commands) {`
`402`		`- channelWrite(command).addListener(RetryListener.newInstance(this, command));`
	`401`	`+ channelWrite(channel, command).addListener(RetryListener.newInstance(this, command));`
`403`	`402`	`}`
`404`	`403`	`}`
`405`	`404`
`406`		`- channelFlush();`
	`405`	`+ channelFlush(channel);`
`407`	`406`	`}`
`408`	`407`
`409`		`- private void channelFlush() {`
	`408`	`+ private void channelFlush(Channel channel) {`
`410`	`409`
`411`	`410`	`if (debugEnabled) {`
`412`	`411`	`logger.debug("{} write() channelFlush", logPrefix());`
`@@ -415,7 +414,7 @@ private void channelFlush() {`
`415`	`414`	`channel.flush();`
`416`	`415`	`}`
`417`	`416`
`418`		`- private ChannelFuture channelWrite(RedisCommand<?, ?, ?> command) {`
	`417`	`+ private ChannelFuture channelWrite(Channel channel, RedisCommand<?, ?, ?> command) {`
`419`	`418`
`420`	`419`	`if (debugEnabled) {`
`421`	`420`	`logger.debug("{} write() channelWrite command {}", logPrefix(), command);`
`@@ -424,7 +423,7 @@ private ChannelFuture channelWrite(RedisCommand<?, ?, ?> command) {`
`424`	`423`	`return channel.write(command);`
`425`	`424`	`}`
`426`	`425`
`427`		`- private ChannelFuture channelWriteAndFlush(RedisCommand<?, ?, ?> command) {`
	`426`	`+ private ChannelFuture channelWriteAndFlush(Channel channel, RedisCommand<?, ?, ?> command) {`
`428`	`427`
`429`	`428`	`if (debugEnabled) {`
`430`	`429`	`logger.debug("{} write() writeAndFlush command {}", logPrefix(), command);`
`@@ -437,7 +436,6 @@ private ChannelFuture channelWriteAndFlush(RedisCommand<?, ?, ?> command) {`
`437`	`436`	`public void notifyChannelActive(Channel channel) {`
`438`	`437`
`439`	`438`	`this.logPrefix = null;`
`440`		`- this.channel = channel;`
`441`	`439`	`this.connectionError = null;`
`442`	`440`
`443`	`441`	`if (isClosed()) {`
`@@ -452,6 +450,7 @@ public void notifyChannelActive(Channel channel) {`
`452`	`450`	`}`
`453`	`451`
`454`	`452`	`sharedLock.doExclusive(() -> {`
	`453`	`+ this.channel = channel;`
`455`	`454`
`456`	`455`	`try {`
`457`	`456`	`// Move queued commands to buffer before issuing any commands because of connection activation.`
`@@ -474,7 +473,7 @@ public void notifyChannelActive(Channel channel) {`
`474`	`473`	`inActivation = false;`
`475`	`474`	`}`
`476`	`475`
`477`		`- flushCommands(disconnectedBuffer);`
	`476`	`+ flushCommands(channel, disconnectedBuffer);`
`478`	`477`	`} catch (Exception e) {`
`479`	`478`
`480`	`479`	`if (debugEnabled) {`
`@@ -527,7 +526,7 @@ public void notifyException(Throwable t) {`
`527`	`526`	`doExclusive(this::drainCommands).forEach(cmd -> cmd.completeExceptionally(t));`
`528`	`527`	`}`
`529`	`528`
`530`		`- if (!isConnected()) {`
	`529`	`+ if (!isConnected(this.channel)) {`
`531`	`530`	`connectionError = t;`
`532`	`531`	`}`
`533`	`532`	`}`
`@@ -540,16 +539,16 @@ public void registerConnectionWatchdog(ConnectionWatchdog connectionWatchdog) {`
`540`	`539`	`@Override`
`541`	`540`	`@SuppressWarnings({ "rawtypes", "unchecked" })`
`542`	`541`	`public void flushCommands() {`
`543`		`- flushCommands(commandBuffer);`
	`542`	`+ flushCommands(this.channel, commandBuffer);`
`544`	`543`	`}`
`545`	`544`
`546`		`- private void flushCommands(Queue<RedisCommand<?, ?, ?>> queue) {`
	`545`	`+ private void flushCommands(Channel channel, Queue<RedisCommand<?, ?, ?>> queue) {`
`547`	`546`
`548`	`547`	`if (debugEnabled) {`
`549`	`548`	`logger.debug("{} flushCommands()", logPrefix());`
`550`	`549`	`}`
`551`	`550`
`552`		`- if (isConnected()) {`
	`551`	`+ if (isConnected(channel)) {`
`553`	`552`
`554`	`553`	`List<RedisCommand<?, ?, ?>> commands = sharedLock.doExclusive(() -> {`
`555`	`554`
`@@ -565,7 +564,7 @@ private void flushCommands(Queue<RedisCommand<?, ?, ?>> queue) {`
`565`	`564`	`}`
`566`	`565`
`567`	`566`	`if (!commands.isEmpty()) {`
`568`		`- writeToChannelAndFlush(commands);`
	`567`	`+ writeToChannelAndFlush(channel, commands);`
`569`	`568`	`}`
`570`	`569`	`}`
`571`	`570`	`}`
`@@ -628,10 +627,10 @@ public void disconnect() {`
`628`	`627`
`629`	`628`	`private Channel getOpenChannel() {`
`630`	`629`
`631`		`- Channel currentChannel = this.channel;`
	`630`	`+ Channel channel = this.channel;`
`632`	`631`
`633`		`- if (currentChannel != null) {`
`634`		`- return currentChannel;`
	`632`	`+ if (channel != null /* && channel.isOpen() is this deliberately omitted? */) {`
	`633`	`+ return channel;`
`635`	`634`	`}`
`636`	`635`
`637`	`636`	`return null;`
`@@ -648,6 +647,7 @@ public void reset() {`
`648`	`647`	`logger.debug("{} reset()", logPrefix());`
`649`	`648`	`}`
`650`	`649`
	`650`	`+ Channel channel = this.channel;`
`651`	`651`	`if (channel != null) {`
`652`	`652`	`channel.pipeline().fireUserEventTriggered(new ConnectionEvents.Reset());`
`653`	`653`	`}`
`@@ -720,9 +720,7 @@ public void notifyDrainQueuedCommands(HasQueuedCommands queuedCommands) {`
`720`	`720`	`}`
`721`	`721`	`}`
`722`	`722`
`723`		`- if (isConnected()) {`
`724`		`- flushCommands(disconnectedBuffer);`
`725`		`- }`
	`723`	`+ flushCommands(this.channel, disconnectedBuffer);`
`726`	`724`	`});`
`727`	`725`	`}`
`728`	`726`
`@@ -787,9 +785,7 @@ private void cancelCommands(String message, Iterable<? extends RedisCommand<?, ?`
`787`	`785`	`}`
`788`	`786`	`}`
`789`	`787`
`790`		`- private boolean isConnected() {`
`791`		`-`
`792`		`- Channel channel = this.channel;`
	`788`	`+ private boolean isConnected(Channel channel) {`
`793`	`789`	`return channel != null && channel.isActive();`
`794`	`790`	`}`
`795`	`791`