RabbitMqIO { public static Write write() { return new AutoValue_RabbitMqIO_Write.Builder() .setExchangeDeclare(false) .setQueueDeclare(false) .build(); } private  RabbitMqIO(); static Read read(); static Write write();  }
@Test public void testWriteQueue() throws Exception { final int maxNumRecords = 1000; List<RabbitMqMessage> data = RabbitMqTestUtils.generateRecords(maxNumRecords).stream() .map(RabbitMqMessage::new) .collect(Collectors.toList()); p.apply(Create.of(data)) .apply( RabbitMqIO.write().withUri("amqp: ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setUri("amqp: Connection connection = null; Channel channel = null; try { connection = connectionFactory.newConnection(); channel = connection.createChannel(); channel.queueDeclare("TEST", true, false, false, null); RabbitMqTestUtils.TestConsumer consumer = new RabbitMqTestUtils.TestConsumer(channel); channel.basicConsume("TEST", true, consumer); p.run(); while (consumer.getReceived().size() < maxNumRecords) { Thread.sleep(500); } assertEquals(maxNumRecords, consumer.getReceived().size()); for (int i = 0; i < maxNumRecords; i++) { assertTrue(consumer.getReceived().contains("Test " + i)); } } finally { if (channel != null) { channel.close(); } if (connection != null) { connection.close(); } } }
@Test public void testWriteExchange() throws Exception { final int maxNumRecords = 1000; List<RabbitMqMessage> data = RabbitMqTestUtils.generateRecords(maxNumRecords).stream() .map(RabbitMqMessage::new) .collect(Collectors.toList()); p.apply(Create.of(data)) .apply( RabbitMqIO.write() .withUri("amqp: .withExchange("WRITE", "fanout")); ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setUri("amqp: Connection connection = null; Channel channel = null; try { connection = connectionFactory.newConnection(); channel = connection.createChannel(); channel.exchangeDeclare("WRITE", "fanout"); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, "WRITE", ""); RabbitMqTestUtils.TestConsumer consumer = new RabbitMqTestUtils.TestConsumer(channel); channel.basicConsume(queueName, true, consumer); p.run(); while (consumer.getReceived().size() < maxNumRecords) { Thread.sleep(500); } assertEquals(maxNumRecords, consumer.getReceived().size()); for (int i = 0; i < maxNumRecords; i++) { assertTrue(consumer.getReceived().contains("Test " + i)); } } finally { if (channel != null) { channel.close(); } if (connection != null) { connection.close(); } } }
@Test public void testWriteQueue() throws Exception { final int maxNumRecords = 1000; List<RabbitMqMessage> data = RabbitMqTestUtils.generateRecords(maxNumRecords).stream() .map(RabbitMqMessage::new) .collect(Collectors.toList()); p.apply(Create.of(data)) .apply( RabbitMqIO.write().withUri("amqp: final List<String> received = new ArrayList<>(); ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setUri("amqp: Connection connection = null; Channel channel = null; try { connection = connectionFactory.newConnection(); channel = connection.createChannel(); channel.queueDeclare("TEST", true, false, false, null); Consumer consumer = new RabbitMqTestUtils.TestConsumer(channel, received); channel.basicConsume("TEST", true, consumer); p.run(); while (received.size() < maxNumRecords) { Thread.sleep(500); } assertEquals(maxNumRecords, received.size()); for (int i = 0; i < maxNumRecords; i++) { assertTrue(received.contains("Test " + i)); } } finally { if (channel != null) { channel.close(); } if (connection != null) { connection.close(); } } }
@Test public void testWriteExchange() throws Exception { final int maxNumRecords = 1000; List<RabbitMqMessage> data = RabbitMqTestUtils.generateRecords(maxNumRecords).stream() .map(RabbitMqMessage::new) .collect(Collectors.toList()); p.apply(Create.of(data)) .apply( RabbitMqIO.write() .withUri("amqp: .withExchange("WRITE", "fanout")); final List<String> received = new ArrayList<>(); ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setUri("amqp: Connection connection = null; Channel channel = null; try { connection = connectionFactory.newConnection(); channel = connection.createChannel(); channel.exchangeDeclare("WRITE", "fanout"); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, "WRITE", ""); Consumer consumer = new RabbitMqTestUtils.TestConsumer(channel, received); channel.basicConsume(queueName, true, consumer); p.run(); while (received.size() < maxNumRecords) { Thread.sleep(500); } assertEquals(maxNumRecords, received.size()); for (int i = 0; i < maxNumRecords; i++) { assertTrue(received.contains("Test " + i)); } } finally { if (channel != null) { channel.close(); } if (connection != null) { connection.close(); } } }