GH-65: Add polled-consumer Sample

Resolves https://github.com/spring-cloud/spring-cloud-stream-samples/issues/65

Polishing - PR Comments

- add docker-compose
- change runner to process multiple messages

processor-samples/polled-consumer/src/main/java/demo/PolledConsumerApplication.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package demo;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.Input;
+import org.springframework.cloud.stream.annotation.Output;
+import org.springframework.cloud.stream.binder.PollableMessageSource;
+import org.springframework.context.annotation.Bean;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.MessageBuilder;
+
+/**
+ * Sample app demonstrating a polled consumer where the application can
+ * control the rate at which messages are retrieved.
+ *
+ * @author Gary Russell
+ *
+ */
+@SpringBootApplication
+@EnableBinding(PolledConsumerApplication.PolledProcessor.class)
+public class PolledConsumerApplication {
+
+	public static final ExecutorService exec = Executors.newSingleThreadExecutor();
+
+	public static void main(String[] args) {
+		SpringApplication.run(PolledConsumerApplication.class, args);
+	}
+
+	@Bean
+	public ApplicationRunner runner(PollableMessageSource input, MessageChannel output) {
+		return args -> {
+			System.out.println("Send some messages to topic polledConsumerIn and receive from polledConsumerOut");
+			System.out.println("Messages will be processed one per second");
+			exec.execute(() -> {
+				boolean result = false;
+				while (true) {
+					// this is where we poll for a message, process it, and send a new one
+					result = input.poll(m -> {
+						String payload = (String) m.getPayload();
+						System.out.println("Received: " + payload);
+						output.send(MessageBuilder.withPayload(payload.toUpperCase())
+							.copyHeaders(m.getHeaders())
+							.build());
+					}, new ParameterizedTypeReference<String>() { });
+
+					try {
+						Thread.sleep(1_000);
+					}
+					catch (InterruptedException e) {
+						Thread.currentThread().interrupt();
+						break;
+					}
+					if (result) {
+						System.out.println("Success");
+					}
+				}
+			});
+		};
+	}
+
+	public interface PolledProcessor {
+
+		@Input
+		PollableMessageSource input();
+
+		@Output
+		MessageChannel output();
+
+	}
+
+}

processor-samples/polled-consumer/src/main/resources/application.properties

@@ -0,0 +1,5 @@
+# Binding properties
+spring.cloud.stream.bindings.output.destination=polledConsumerOut
+spring.cloud.stream.bindings.input.destination=polledConsumerIn
+spring.cloud.stream.bindings.input.group=polledConsumerApplication
+spring.cloud.stream.bindings.input.content-type=text/plain

processor-samples/polled-consumer/src/test/java/demo/PolledConsumerApplicationTests.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package demo;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * Test class demonstrating how to use an embedded kafka service with the
+ * kafka binder to test the {@link PolledConsumerApplication}.
+ *
+ * @author Gary Russell
+ *
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class PolledConsumerApplicationTests {
+
+	@ClassRule
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1);
+
+	@Value("${spring.cloud.stream.bindings.input.destination}")
+	private String inputTopic;
+
+	@Value("${spring.cloud.stream.bindings.output.destination}")
+	private String outputTopic;
+
+	@BeforeClass
+	public static void setup() {
+		System.setProperty("spring.kafka.bootstrap-servers", embeddedKafka.getBrokersAsString());
+	}
+
+	@Test
+	public void testSendReceive() throws Exception {
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		senderProps.put("key.serializer", ByteArraySerializer.class);
+		senderProps.put("value.serializer", ByteArraySerializer.class);
+		DefaultKafkaProducerFactory<byte[], byte[]> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<byte[], byte[]> template = new KafkaTemplate<>(pf, true);
+		template.setDefaultTopic(inputTopic);
+		template.sendDefault("foo".getBytes());
+
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testGroup", "false", embeddedKafka);
+		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		consumerProps.put("key.deserializer", ByteArrayDeserializer.class);
+		consumerProps.put("value.deserializer", ByteArrayDeserializer.class);
+		DefaultKafkaConsumerFactory<byte[], byte[]> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
+
+		Consumer<byte[], byte[]> consumer = cf.createConsumer();
+		consumer.subscribe(Collections.singleton(this.outputTopic));
+		ConsumerRecords<byte[], byte[]> records = consumer.poll(10_000);
+		consumer.commitSync();
+
+		assertThat(records.count()).isEqualTo(1);
+		assertThat(new String(records.iterator().next().value())).isEqualTo("FOO");
+		consumer.close();
+		pf.destroy();
+
+		PolledConsumerApplication.exec.shutdownNow();
+	}
+
+}