@Test(expected = IllegalStateException.class) public void failPublishIfChannelFails() throws Exception { TestConnection conn = new TestConnection(null, null); try (AMQPPublisher sender = new AMQPPublisher(conn, mock(ComponentLog.class))) { ((TestChannel) conn.createChannel()).corruptChannel(); sender.publish("oleg".getBytes(), null, "foo", ""); } }
void publish(byte[] bytes, BasicProperties properties, String routingKey, String exchange) { this.validateStringProperty("routingKey", routingKey); exchange = exchange == null ? "" : exchange.trim(); if (exchange.length() == 0) { processLog.info("The 'exchangeName' is not specified. Messages will be sent to default exchange"); } processLog.info("Successfully connected AMQPPublisher to " + this.connectionString + " and '" + exchange + "' exchange with '" + routingKey + "' as a routing key."); if (this.channel.isOpen()) { try { this.channel.basicPublish(exchange, routingKey, true, properties, bytes); } catch (Exception e) { throw new IllegalStateException("Failed to publish to '" + exchange + "' with '" + routingKey + "'.", e); } } else { throw new IllegalStateException("This instance of AMQPPublisher is invalid since " + "its publishingChannel is closed"); } }
AMQPPublisher extends AMQPWorker { void publish(byte[] bytes, BasicProperties properties, String routingKey, String exchange) { this.validateStringProperty("routingKey", routingKey); exchange = exchange == null ? "" : exchange.trim(); if (exchange.length() == 0) { processLog.info("The 'exchangeName' is not specified. Messages will be sent to default exchange"); } processLog.info("Successfully connected AMQPPublisher to " + this.connectionString + " and '" + exchange + "' exchange with '" + routingKey + "' as a routing key."); if (this.channel.isOpen()) { try { this.channel.basicPublish(exchange, routingKey, true, properties, bytes); } catch (Exception e) { throw new IllegalStateException("Failed to publish to '" + exchange + "' with '" + routingKey + "'.", e); } } else { throw new IllegalStateException("This instance of AMQPPublisher is invalid since " + "its publishingChannel is closed"); } } }
AMQPPublisher extends AMQPWorker { void publish(byte[] bytes, BasicProperties properties, String routingKey, String exchange) { this.validateStringProperty("routingKey", routingKey); exchange = exchange == null ? "" : exchange.trim(); if (exchange.length() == 0) { processLog.info("The 'exchangeName' is not specified. Messages will be sent to default exchange"); } processLog.info("Successfully connected AMQPPublisher to " + this.connectionString + " and '" + exchange + "' exchange with '" + routingKey + "' as a routing key."); if (this.channel.isOpen()) { try { this.channel.basicPublish(exchange, routingKey, true, properties, bytes); } catch (Exception e) { throw new IllegalStateException("Failed to publish to '" + exchange + "' with '" + routingKey + "'.", e); } } else { throw new IllegalStateException("This instance of AMQPPublisher is invalid since " + "its publishingChannel is closed"); } } AMQPPublisher(Connection connection, ComponentLog processLog); }
AMQPPublisher extends AMQPWorker { void publish(byte[] bytes, BasicProperties properties, String routingKey, String exchange) { this.validateStringProperty("routingKey", routingKey); exchange = exchange == null ? "" : exchange.trim(); if (exchange.length() == 0) { processLog.info("The 'exchangeName' is not specified. Messages will be sent to default exchange"); } processLog.info("Successfully connected AMQPPublisher to " + this.connectionString + " and '" + exchange + "' exchange with '" + routingKey + "' as a routing key."); if (this.channel.isOpen()) { try { this.channel.basicPublish(exchange, routingKey, true, properties, bytes); } catch (Exception e) { throw new IllegalStateException("Failed to publish to '" + exchange + "' with '" + routingKey + "'.", e); } } else { throw new IllegalStateException("This instance of AMQPPublisher is invalid since " + "its publishingChannel is closed"); } } AMQPPublisher(Connection connection, ComponentLog processLog); @Override String toString(); }
AMQPPublisher extends AMQPWorker { void publish(byte[] bytes, BasicProperties properties, String routingKey, String exchange) { this.validateStringProperty("routingKey", routingKey); exchange = exchange == null ? "" : exchange.trim(); if (exchange.length() == 0) { processLog.info("The 'exchangeName' is not specified. Messages will be sent to default exchange"); } processLog.info("Successfully connected AMQPPublisher to " + this.connectionString + " and '" + exchange + "' exchange with '" + routingKey + "' as a routing key."); if (this.channel.isOpen()) { try { this.channel.basicPublish(exchange, routingKey, true, properties, bytes); } catch (Exception e) { throw new IllegalStateException("Failed to publish to '" + exchange + "' with '" + routingKey + "'.", e); } } else { throw new IllegalStateException("This instance of AMQPPublisher is invalid since " + "its publishingChannel is closed"); } } AMQPPublisher(Connection connection, ComponentLog processLog); @Override String toString();  }