package com.hubspot.smtp.client;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.smtp.SmtpResponse;

class ResponseHandler extends ChannelInboundHandlerAdapter {
  private static final Logger LOG = LoggerFactory.getLogger(ResponseHandler.class);

  private final AtomicReference<ResponseCollector> responseCollector = new AtomicReference<>();

  CompletableFuture<SmtpResponse[]> createResponseFuture(int expectedResponses, Supplier<String> debugStringSupplier) {
    ResponseCollector collector = new ResponseCollector(expectedResponses, debugStringSupplier);

    boolean success = responseCollector.compareAndSet(null, collector);
    if (!success) {
      ResponseCollector previousCollector = this.responseCollector.get();
      if (previousCollector == null) {
        return createResponseFuture(expectedResponses, debugStringSupplier);
      }

      throw new IllegalStateException(String.format("Cannot wait for a response to [%s] because we're still waiting for a response to [%s]",
          collector.getDebugString(), previousCollector.getDebugString()));
    }

    // although the future field may have been written in another thread,
    // the compareAndSet call above has volatile semantics and
    // ensures the write will be visible
    return collector.getFuture();
  }

  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof SmtpResponse) {
      ResponseCollector collector = responseCollector.get();

      if (collector == null) {
        LOG.warn("Unexpected response received: {}", msg);
      } else {
        boolean complete = collector.addResponse((SmtpResponse) msg);
        if (complete) {
          // because only the event loop code sets this field when it is non-null,
          // and because channelRead is always run in the same thread, we can
          // be sure this value hasn't changed since we read it earlier in this method
          responseCollector.set(null);
          collector.complete();
        }
      }
    }

    ctx.fireChannelRead(msg);
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    ResponseCollector collector = responseCollector.getAndSet(null);
    if (collector != null) {
      collector.completeExceptionally(cause);
    }
  }
}

@@ -288,24 +288,30 @@
 tty.channel.
+Simple
 ChannelInbou
@@ -319,23 +319,16 @@
 dHandler
-Adapter
 ;%0Aimport
@@ -401,16 +401,22 @@
 extends 
+Simple
 ChannelI
@@ -432,15 +432,22 @@
 dler
-Adapter
+%3CSmtpResponse%3E
  %7B%0A 
@@ -1588,29 +1588,32 @@
 Override%0A  p
-ublic
+rotected
  void channe
@@ -1617,16 +1617,17 @@
 nnelRead
+0
 (Channel
@@ -1650,14 +1650,20 @@
 tx, 
-Object
+SmtpResponse
  msg
@@ -1687,49 +1687,8 @@
 n %7B%0A
-    if (msg instanceof SmtpResponse) %7B%0A  
     
@@ -1743,26 +1743,24 @@
 get();%0A%0A    
-  
 if (collecto
@@ -1768,26 +1768,24 @@
  == null) %7B%0A
-  
       LOG.wa
@@ -1833,18 +1833,16 @@
 g);%0A    
-  
 %7D else %7B
@@ -1842,18 +1842,16 @@
  else %7B%0A
-  
       bo
@@ -1893,31 +1893,14 @@
 nse(
-(SmtpResponse) 
 msg);%0A
-  
     
@@ -1913,26 +1913,24 @@
 complete) %7B%0A
-  
         // b
@@ -2006,18 +2006,16 @@
         
-  
 // and b
@@ -2074,18 +2074,16 @@
  we can%0A
-  
         
@@ -2167,18 +2167,16 @@
         
-  
 response
@@ -2196,34 +2196,32 @@
 (null);%0A        
-  
 collector.comple
@@ -2236,56 +2236,15 @@
     
- 
- 
 %7D%0A    
-  %7D%0A    %7D%0A%0A    ctx.fireChannelRead(msg);
+%7D
 %0A  %7D

Extend SimpleChannelInboundHandler for simpler method signature