SqsIO { public static Read read() { return new AutoValue_SqsIO_Read.Builder().setMaxNumRecords(Long.MAX_VALUE).build(); } private  SqsIO(); static Read read(); static Write write();  }
@Test public void testRead() { final AmazonSQS client = embeddedSqsRestServer.getClient(); final String queueUrl = embeddedSqsRestServer.getQueueUrl(); final PCollection<Message> output = pipeline.apply(SqsIO.read().withQueueUrl(queueUrl).withMaxNumRecords(100)); PAssert.thatSingleton(output.apply(Count.globally())).isEqualTo(100L); for (int i = 0; i < 100; i++) { client.sendMessage(queueUrl, "This is a test"); } pipeline.run(); }
@Test public void testRead() { final SqsClient client = EmbeddedSqsServer.getClient(); final String queueUrl = EmbeddedSqsServer.getQueueUrl(); final PCollection<SqsMessage> output = pipeline.apply( SqsIO.read() .withSqsClientProvider(SqsClientProviderMock.of(client)) .withQueueUrl(queueUrl) .withMaxNumRecords(100)); PAssert.thatSingleton(output.apply(Count.globally())).isEqualTo(100L); for (int i = 0; i < 100; i++) { SendMessageRequest sendMessageRequest = SendMessageRequest.builder().queueUrl(queueUrl).messageBody("This is a test").build(); client.sendMessage(sendMessageRequest); } pipeline.run(); }