Java 类com.rabbitmq.client.ShutdownListener 实例源码

private void initRabbitMQ() throws IOException {
    Server.LOGGER.info("Initialization of the Notifications channel");
    mRabbitMQManager.getChannel().addShutdownListener(new ShutdownListener() {

        @Override
        public void shutdownCompleted(ShutdownSignalException cause) {
            cause.printStackTrace();
        }
    });
    mRabbitMQManager.getChannel().exchangeDeclare("Pub", BuiltinExchangeType.FANOUT, true);
    String queueName = mRabbitMQManager.getChannel().queueDeclare().getQueue();
    mRabbitMQManager.getChannel().queueBind(queueName, "Pub", "");

    mRabbitMQManager.getChannel().basicConsume(queueName, true, new RabbitMQConsumer(this, mRabbitMQManager.getChannel()));
    Server.LOGGER.info("Initialization of the Pub channel done.");
}

/****
 * This method is used to publish the message to RabbitMQ
 * @param routingKey
 * @param msg is Eiffel Event
 * @throws IOException
 */
public void send(String routingKey, String msg) throws IOException {

    Channel channel = giveMeRandomChannel();
    channel.addShutdownListener(new ShutdownListener() {
        public void shutdownCompleted(ShutdownSignalException cause) {
            // Beware that proper synchronization is needed here
            if (cause.isInitiatedByApplication()) {
                log.debug("Shutdown is initiated by application. Ignoring it.");
            } else {
                log.error("Shutdown is NOT initiated by application.");
                log.error(cause.getMessage());
                boolean cliMode = Boolean.getBoolean(PropertiesConfig.CLI_MODE);
                if (cliMode) {
                    System.exit(-3);
                }
            }
        }
    });

    BasicProperties msgProps = MessageProperties.BASIC;
    if (usePersitance)
        msgProps = MessageProperties.PERSISTENT_BASIC;

    channel.basicPublish(exchangeName, routingKey, msgProps, msg.getBytes());
    log.info("Published message with size {} bytes on exchange '{}' with routing key '{}'", msg.getBytes().length,
            exchangeName, routingKey);
}

/**
 * Handles common method invocations.
 */
boolean handleCommonMethods(Object delegate, Method method, Object[] args) throws Throwable {
  if ("abort".equals(method.getName()) || "close".equals(method.getName())) {
    try {
      Reflection.invoke(delegate, method, args);
      return true;
    } finally {
      closed = true;
      afterClosure();
      interruptWaiters();
    }
  } else if ("addShutdownListener".equals(method.getName()) && args[0] != null)
    shutdownListeners.add((ShutdownListener) args[0]);
  else if ("removeShutdownListener".equals(method.getName()) && args[0] != null)
    shutdownListeners.remove((ShutdownListener) args[0]);
  return false;
}

protected synchronized void init() throws RabbitMQTransportException
{
    try
    {
        channel.addShutdownListener(new ShutdownListener()
            {
                @Override
                public void shutdownCompleted(ShutdownSignalException cause)
                {
                    disconnect(cause.getMessage());
                }
            });

        channel.exchangeDeclare(
                exchange.getName(),
                exchange.getType().toString(),
                exchange.isDurable(),
                exchange.isAutoDelete(),
                null
            );
    }
    catch (IOException e)
    {
        String msg = LOGGER.translate("EXCHANGE_CREATE_ERROR", e.getMessage());
        LOGGER.error(msg, e);
        throw new RabbitMQTransportException(msg);
    }
}

private void connect() throws AiravataException {
    try {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setUri(properties.getBrokerUrl());
        connectionFactory.setAutomaticRecoveryEnabled(properties.isAutoRecoveryEnable());
        connection = connectionFactory.newConnection();
        connection.addShutdownListener(new ShutdownListener() {
            public void shutdownCompleted(ShutdownSignalException cause) {
            }
        });
        log.info("connected to rabbitmq: " + connection + " for " + properties.getExchangeName());
        channel = connection.createChannel();
        if (properties.getPrefetchCount() > 0) {
            channel.basicQos(properties.getPrefetchCount());
        }

        if (properties.getExchangeName() != null) {
            channel.exchangeDeclare(properties.getExchangeName(),
                                    properties.getExchangeType(),
                                    true); //durable
        }
    } catch (Exception e) {
        String msg = "RabbitMQ connection issue for exchange : " + properties.getExchangeName();
        log.error(msg);
        throw new AiravataException(msg, e);
    }


}

private void addShutdownListener() {
    connection.addShutdownListener(new ShutdownListener() {
        public void shutdownCompleted(ShutdownSignalException cause) {
            log.error("RabbitMQ connection " + connection + " for " + properties.getExchangeName() + " has been shut down", cause);
        }
    });
}

private void attachToQueue(boolean recover) throws IOException {
    Connection cn = rabbitConnectionFactory.createConnection();
    this.channel = cn.createChannel(false);
    channel.exchangeDeclare(this.topicName(), "topic", true);
    this.queueName = channel.queueDeclare(
            this.topicName() + ".__." + this.queueName()
            , true, false, false, null).getQueue();
    if (recover) {
        channel.basicRecover(true);
    }
    if (!recover) RabbitMqListenerRegistry.registerListener(this, 10000);
    channel.addShutdownListener(new ShutdownListener() {
        @Override
        public void shutdownCompleted(ShutdownSignalException cause) {
            if (!cause.isInitiatedByApplication()) {
                log.error("Channel shutdown: " + cause.getMessage(), cause);
                log.info("Re creating channel..");
                try {
                    attachToQueue(true);
                } catch (IOException ex) {
                    log.error("Couldn't create new channel due to " + ex, ex);
                }
            } else {
                if (log.isDebugEnabled())
                    log.debug("Channel shutdown: " + cause.getMessage());
            }
        }
    });
    this.consumerTag = this.topicName() + ".__." + this.queueName() + UUID.randomUUID().toString();
    registerConsumer(channel, queueName, consumerTag);
}

public ChannelHandler(ConnectionHandler connectionHandler, Channel delegate, Config config) {
  this.connectionHandler = connectionHandler;
  this.delegate = delegate;
  this.config = config;

  ShutdownListener listener = new ChannelShutdownListener();
  shutdownListeners.add(listener);
  delegate.addShutdownListener(listener);
}

@Override
public Connection create() throws Exception {
    ConnectionFactory factory = null;
    Connection connection = null;
    do {
        try {
            factory = pickOne();
            connection = factory.newConnection();
        } catch (Exception e) {
            logger.error("fail to create new connection from factory: [" + factory.getHost() + ":"
                    + factory.getPort() + "], kicking this one out and retry...");
            kick(factory.getHost() + ":" + factory.getPort());
        }
    } while (connection == null
            && ConnectionFactoryManager.getInstance().getAvailableFactories().keySet().size() > 0);
    if (connection == null) {
        throw new Exception("fail to get new connection. no hosts left to use.");
    }
    /* ADD CONNECTION & CHANNEL CONNECTION LISTENER */
    connection.addShutdownListener(new ShutdownListener() {
        public void shutdownCompleted(ShutdownSignalException cause) {
            String hardError = "";
            String applInit = "";
            if (cause.isHardError()) {
                hardError = "connection";
            } else {
                hardError = "channel";
            }

            if (cause.isInitiatedByApplication()) {
                applInit = "application";
            } else {
                applInit = "broker";
            }
            logger.warn("Connectivity to MQ has failed.  It was caused by " + applInit + " at the " + hardError
                    + " level.  Reason received " + cause.getReason());
        }
    });

    ((Recoverable) connection).addRecoveryListener(new RecoveryListener() {
        public void handleRecovery(Recoverable recoverable) {
            if (recoverable instanceof Connection) {
                logger.info("Connection was recovered.");
            } else if (recoverable instanceof Channel) {
                int channelNumber = ((Channel) recoverable).getChannelNumber();
                logger.info("Connection to channel #" + channelNumber + " was recovered.");
            }
        }

        public void handleRecoveryStarted(Recoverable arg0) {
        }
    });
    /* ADD CONNECTION & CHANNEL CONNECTION LISTENER */
    logger.info(
            "new connection was establesed...from host <- [" + factory.getHost() + ":" + factory.getPort() + "]");
    return connection;
}

@Override
public void addShutdownListener(ShutdownListener listener) {
  channel.addShutdownListener(listener);
}

@Override
public void removeShutdownListener(ShutdownListener listener) {
  channel.removeShutdownListener(listener);
}

@Override
public void addShutdownListener(ShutdownListener listener) {
  this.mc.getUnderlyingConnection().addShutdownListener(listener);
}

@Override
public void removeShutdownListener(ShutdownListener listener) {
  this.mc.getUnderlyingConnection().removeShutdownListener(listener);
}

@Override
public void addShutdownListener(ShutdownListener listener) {
  this.mc.getUnderlyingConnection().addShutdownListener(listener);
}

@Override
public void removeShutdownListener(ShutdownListener listener) {
  this.mc.getUnderlyingConnection().removeShutdownListener(listener);
}