@Test public void validateSuccessfullPublishingAndRouting() throws Exception { Map<String, List<String>> routingMap = new HashMap<>(); routingMap.put("key1", Arrays.asList("queue1", "queue2")); Map<String, String> exchangeToRoutingKeymap = new HashMap<>(); exchangeToRoutingKeymap.put("myExchange", "key1"); Connection connection = new TestConnection(exchangeToRoutingKeymap, routingMap); try (AMQPPublisher sender = new AMQPPublisher(connection, mock(ComponentLog.class))) { sender.publish("hello".getBytes(), null, "key1", "myExchange"); Thread.sleep(200); } assertNotNull(connection.createChannel().basicGet("queue1", true)); assertNotNull(connection.createChannel().basicGet("queue2", true)); connection.close(); }
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();  }