// Path: beetle-core/src/main/java/com/xing/beetle/util/ExceptionSupport.java
// @FunctionalInterface
// public interface Function<T, R> extends java.util.function.Function<T, R> {
// 
//   @Override
//   default R apply(T t) {
//     try {
//       return applyChecked(t);
//     } catch (Throwable e) {
//       return ExceptionSupport.sneakyThrow(e);
//     }
//   }
// 
//   R applyChecked(T t) throws Throwable;
// }


import static java.util.Objects.requireNonNull;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;
import com.xing.beetle.util.ExceptionSupport.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
package com.xing.beetle.amqp;





/** BeetleConnection aggregates connections to one or more brokers to one virtual connection. */
public class BeetleConnection implements DefaultConnection.Decorator, ShutdownListener {

  private static final Logger LOGGER = LoggerFactory.getLogger(BeetleConnection.class);

  private final List<? extends Connection> delegates;
  private final BeetleAmqpConfiguration configuration;
  private final Set<ShutdownListener> shutdownListeners;

  /**
   * Build a new BeetleConnection that aggregfates the given connections.
   *
   * @param connections Wrapped AMQP connections.
   */
  public BeetleConnection(List<Connection> connections, BeetleAmqpConfiguration configuration) {
    this.delegates = new ArrayList<>(connections);
    this.configuration = configuration;
    this.shutdownListeners = new HashSet<>();
    connections.forEach(c -> c.addShutdownListener(this));
  }

  @Override
  public void addShutdownListener(ShutdownListener listener) {
    shutdownListeners.add(requireNonNull(listener));
  }

  @Override
  public Channel createChannel(int channelNumber) throws IOException {
    List<Channel> channels = new ArrayList<>();
    for (Connection connection : delegates) {
      channels.add(
          channelNumber >= 0
              ? connection.createChannel(channelNumber)
              : connection.createChannel());
    }
    return new BeetleChannel(channels, configuration);
  }

  @Override
  public <R> R delegateMap(Function<Connection, ? extends R> fn) {
    return delegates.stream().map(fn).reduce(null, (r1, r2) -> r1 != null ? r1 : r2);
  }

// Path: beetle-core/src/main/java/com/xing/beetle/util/ExceptionSupport.java
// @FunctionalInterface
// public interface Function<T, R> extends java.util.function.Function<T, R> {
// 
//   @Override
//   default R apply(T t) {
//     try {
//       return applyChecked(t);
//     } catch (Throwable e) {
//       return ExceptionSupport.sneakyThrow(e);
//     }
//   }
// 
//   R applyChecked(T t) throws Throwable;
// }


// Path: beetle-core/src/main/java/com/xing/beetle/amqp/BeetleConnection.java
import static java.util.Objects.requireNonNull;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;
import com.xing.beetle.util.ExceptionSupport.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

package com.xing.beetle.amqp;





/** BeetleConnection aggregates connections to one or more brokers to one virtual connection. */
public class BeetleConnection implements DefaultConnection.Decorator, ShutdownListener {

  private static final Logger LOGGER = LoggerFactory.getLogger(BeetleConnection.class);

  private final List<? extends Connection> delegates;
  private final BeetleAmqpConfiguration configuration;
  private final Set<ShutdownListener> shutdownListeners;

  /**
   * Build a new BeetleConnection that aggregfates the given connections.
   *
   * @param connections Wrapped AMQP connections.
   */
  public BeetleConnection(List<Connection> connections, BeetleAmqpConfiguration configuration) {
    this.delegates = new ArrayList<>(connections);
    this.configuration = configuration;
    this.shutdownListeners = new HashSet<>();
    connections.forEach(c -> c.addShutdownListener(this));
  }

  @Override
  public void addShutdownListener(ShutdownListener listener) {
    shutdownListeners.add(requireNonNull(listener));
  }

  @Override
  public Channel createChannel(int channelNumber) throws IOException {
    List<Channel> channels = new ArrayList<>();
    for (Connection connection : delegates) {
      channels.add(
          channelNumber >= 0
              ? connection.createChannel(channelNumber)
              : connection.createChannel());
    }
    return new BeetleChannel(channels, configuration);
  }

  @Override
  public <R> R delegateMap(Function<Connection, ? extends R> fn) {
    return delegates.stream().map(fn).reduce(null, (r1, r2) -> r1 != null ? r1 : r2);
  }

  @Override