// Path: influent-transport/src/main/java/influent/exception/InfluentIOException.java
// public final class InfluentIOException extends RuntimeException {
//   /**
//    * Constructs a new {@code InfluentIOException}.
//    *
//    * @param message the error message
//    * @param cause the cause
//    */
//   public InfluentIOException(final String message, final Throwable cause) {
//     super(message, cause);
//   }
// 
//   /**
//    * Constructs a new {@code InfluentIOException}.
//    *
//    * @param message the error message
//    */
//   public InfluentIOException(final String message) {
//     super(message);
//   }
// 
//   /** Constructs a new {@code InfluentIOException}. */
//   public InfluentIOException() {
//     super();
//   }
// }
// 
// Path: influent-transport/src/main/java/influent/internal/util/Exceptions.java
// public final class Exceptions {
//   /**
//    * A callable block.
//    *
//    * <p>This is the same as {@code Runnable} except that {@code Block#run} may throw some {@code
//    * Exception}.
//    */
//   @FunctionalInterface
//   public interface Block {
//     void run() throws Exception;
//   }
// 
//   /** Boolean specialized callable. */
//   @FunctionalInterface
//   public interface BooleanCallable {
//     boolean call() throws Exception;
//   }
// 
//   private static final Logger logger = LoggerFactory.getLogger(Exceptions.class);
// 
//   private Exceptions() {
//     throw new AssertionError();
//   }
// 
//   /**
//    * Executes the given processing and discards the error when some error occurs.
//    *
//    * @param f the processing
//    * @param messageOnError the error message to be logged on an error
//    */
//   public static void ignore(final Block f, final String messageOnError) {
//     try {
//       f.run();
//     } catch (final Exception e) {
//       logger.error(messageOnError, e);
//     }
//   }
// 
//   /**
//    * Executes the given processing and return null if the processing throws an exception.
//    *
//    * @param f the processing
//    * @param <T> the type of the {@code Callable}
//    * @return the return value of {@code f} or null when {@code f} fails
//    */
//   public static <T> T orNull(final Callable<T> f) {
//     try {
//       return f.call();
//     } catch (final Exception e) {
//       return null;
//     }
//   }
// 
//   /**
//    * Executes the given processing and return false if the processing throws an exception.
//    *
//    * @param f the processing
//    * @return true if {@code f} returns true
//    */
//   public static boolean orFalse(final BooleanCallable f) {
//     try {
//       return f.call();
//     } catch (final Exception e) {
//       return false;
//     }
//   }
// }


import influent.exception.InfluentIOException;
import influent.internal.util.Exceptions;
import java.io.IOException;
import java.net.SocketAddress;
import java.net.SocketOption;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.NotYetConnectedException;
import java.nio.channels.SocketChannel;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/*
 * Copyright 2016 okumin
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package influent.internal.nio;


/** A non-blocking {@code SocketChannel}. */
public final class NioTcpPlaintextChannel implements NioTcpChannel {
  private static final Logger logger = LoggerFactory.getLogger(NioTcpPlaintextChannel.class);

  private final SocketChannel channel;
  private final NioEventLoop eventLoop;
  private final SocketAddress remoteAddress;

  final NioSelectionKey key = NioSelectionKey.create();

  NioTcpPlaintextChannel(
      final SocketChannel channel,
      final NioEventLoop eventLoop,
      final SocketAddress remoteAddress) {
    this.channel = channel;
    this.eventLoop = eventLoop;
    this.remoteAddress = remoteAddress;
  }

  private static SocketAddress getRemoteAddress(final SocketChannel channel) {
    try {
      return channel.getRemoteAddress();
    } catch (final IOException e) {
      // ClosedChannelException is an IOException
      throw new InfluentIOException("SocketChannel#getRemoteAddress failed", e);
    }
  }

  private static <T> void setOption(
      final SocketChannel channel, final SocketOption<T> name, final T value) {
    try {
      channel.setOption(name, value);
    } catch (final UnsupportedOperationException | IllegalArgumentException e) {
      throw new AssertionError(e);
    } catch (final IOException e) {
      // ClosedChannelException is an IOException
      throw new InfluentIOException("SocketChannel#setOption failed", e);
    }
  }

  /**
   * Creates a new {@code NioTcpPlaintextChannel}.
   *
   * @param channel the accepted {@code SocketChannel}
   * @param eventLoop the {@code NioEventLoop}
   * @param tcpConfig the {@code NioTcpConfig}
   * @throws InfluentIOException if some IO error occurs
   */
  public static NioTcpPlaintextChannel open(
      final SocketChannel channel, final NioEventLoop eventLoop, final NioTcpConfig tcpConfig) {
    try {
      final SocketAddress remoteAddress = getRemoteAddress(channel);
      tcpConfig
          .getSendBufferSize()
          .ifPresent(
              (sendBufferSize) ->
                  setOption(channel, StandardSocketOptions.SO_SNDBUF, sendBufferSize));
      setOption(channel, StandardSocketOptions.SO_KEEPALIVE, tcpConfig.getKeepAliveEnabled());
      setOption(channel, StandardSocketOptions.TCP_NODELAY, tcpConfig.getTcpNoDelayEnabled());
      return new NioTcpPlaintextChannel(channel, eventLoop, remoteAddress);
    } catch (final Exception e) {
// Path: influent-transport/src/main/java/influent/exception/InfluentIOException.java
// public final class InfluentIOException extends RuntimeException {
//   /**
//    * Constructs a new {@code InfluentIOException}.
//    *
//    * @param message the error message
//    * @param cause the cause
//    */
//   public InfluentIOException(final String message, final Throwable cause) {
//     super(message, cause);
//   }
// 
//   /**
//    * Constructs a new {@code InfluentIOException}.
//    *
//    * @param message the error message
//    */
//   public InfluentIOException(final String message) {
//     super(message);
//   }
// 
//   /** Constructs a new {@code InfluentIOException}. */
//   public InfluentIOException() {
//     super();
//   }
// }
// 
// Path: influent-transport/src/main/java/influent/internal/util/Exceptions.java
// public final class Exceptions {
//   /**
//    * A callable block.
//    *
//    * <p>This is the same as {@code Runnable} except that {@code Block#run} may throw some {@code
//    * Exception}.
//    */
//   @FunctionalInterface
//   public interface Block {
//     void run() throws Exception;
//   }
// 
//   /** Boolean specialized callable. */
//   @FunctionalInterface
//   public interface BooleanCallable {
//     boolean call() throws Exception;
//   }
// 
//   private static final Logger logger = LoggerFactory.getLogger(Exceptions.class);
// 
//   private Exceptions() {
//     throw new AssertionError();
//   }
// 
//   /**
//    * Executes the given processing and discards the error when some error occurs.
//    *
//    * @param f the processing
//    * @param messageOnError the error message to be logged on an error
//    */
//   public static void ignore(final Block f, final String messageOnError) {
//     try {
//       f.run();
//     } catch (final Exception e) {
//       logger.error(messageOnError, e);
//     }
//   }
// 
//   /**
//    * Executes the given processing and return null if the processing throws an exception.
//    *
//    * @param f the processing
//    * @param <T> the type of the {@code Callable}
//    * @return the return value of {@code f} or null when {@code f} fails
//    */
//   public static <T> T orNull(final Callable<T> f) {
//     try {
//       return f.call();
//     } catch (final Exception e) {
//       return null;
//     }
//   }
// 
//   /**
//    * Executes the given processing and return false if the processing throws an exception.
//    *
//    * @param f the processing
//    * @return true if {@code f} returns true
//    */
//   public static boolean orFalse(final BooleanCallable f) {
//     try {
//       return f.call();
//     } catch (final Exception e) {
//       return false;
//     }
//   }
// }


// Path: influent-transport/src/main/java/influent/internal/nio/NioTcpPlaintextChannel.java
import influent.exception.InfluentIOException;
import influent.internal.util.Exceptions;
import java.io.IOException;
import java.net.SocketAddress;
import java.net.SocketOption;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.NotYetConnectedException;
import java.nio.channels.SocketChannel;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/*
 * Copyright 2016 okumin
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package influent.internal.nio;


/** A non-blocking {@code SocketChannel}. */
public final class NioTcpPlaintextChannel implements NioTcpChannel {
  private static final Logger logger = LoggerFactory.getLogger(NioTcpPlaintextChannel.class);

  private final SocketChannel channel;
  private final NioEventLoop eventLoop;
  private final SocketAddress remoteAddress;

  final NioSelectionKey key = NioSelectionKey.create();

  NioTcpPlaintextChannel(
      final SocketChannel channel,
      final NioEventLoop eventLoop,
      final SocketAddress remoteAddress) {
    this.channel = channel;
    this.eventLoop = eventLoop;
    this.remoteAddress = remoteAddress;
  }

  private static SocketAddress getRemoteAddress(final SocketChannel channel) {
    try {
      return channel.getRemoteAddress();
    } catch (final IOException e) {
      // ClosedChannelException is an IOException
      throw new InfluentIOException("SocketChannel#getRemoteAddress failed", e);
    }
  }

  private static <T> void setOption(
      final SocketChannel channel, final SocketOption<T> name, final T value) {
    try {
      channel.setOption(name, value);
    } catch (final UnsupportedOperationException | IllegalArgumentException e) {
      throw new AssertionError(e);
    } catch (final IOException e) {
      // ClosedChannelException is an IOException
      throw new InfluentIOException("SocketChannel#setOption failed", e);
    }
  }

  /**
   * Creates a new {@code NioTcpPlaintextChannel}.
   *
   * @param channel the accepted {@code SocketChannel}
   * @param eventLoop the {@code NioEventLoop}
   * @param tcpConfig the {@code NioTcpConfig}
   * @throws InfluentIOException if some IO error occurs
   */
  public static NioTcpPlaintextChannel open(
      final SocketChannel channel, final NioEventLoop eventLoop, final NioTcpConfig tcpConfig) {
    try {
      final SocketAddress remoteAddress = getRemoteAddress(channel);
      tcpConfig
          .getSendBufferSize()
          .ifPresent(
              (sendBufferSize) ->
                  setOption(channel, StandardSocketOptions.SO_SNDBUF, sendBufferSize));
      setOption(channel, StandardSocketOptions.SO_KEEPALIVE, tcpConfig.getKeepAliveEnabled());
      setOption(channel, StandardSocketOptions.TCP_NODELAY, tcpConfig.getTcpNoDelayEnabled());
      return new NioTcpPlaintextChannel(channel, eventLoop, remoteAddress);
    } catch (final Exception e) {
      closeChannel(channel, Exceptions.orNull(channel::getRemoteAddress));