package io.dwadden.streams;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AccessLevel;
import lombok.SneakyThrows;
import lombok.experimental.FieldDefaults;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.test.context.junit.jupiter.SpringExtension;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.springframework.integration.test.matcher.PayloadMatcher.hasPayload;

@FieldDefaults(level = AccessLevel.PRIVATE)
@ExtendWith(SpringExtension.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class StreamServiceIntegrationTests {

    static final ObjectMapper objectMapper = new ObjectMapper();

    final StreamService streamService;
    final Source source;
    final MessageCollector collector;

    @Autowired
    StreamServiceIntegrationTests(StreamService streamService, Source source, MessageCollector collector) {
        this.streamService = streamService;
        this.source = source;
        this.collector = collector;
    }

    @SuppressWarnings("unchecked")
    @SneakyThrows({JsonProcessingException.class, InterruptedException.class})
    @DisplayName("should integration test the source")
    @Test
    void ingest() {
        IngestedPayload ingestedPayload = IngestedPayload.builder()
            .type("some-type")
            .payload("some-payload")
            .build();

        streamService.ingestPayload(ingestedPayload);

        String payload = objectMapper.writeValueAsString(ingestedPayload);
        assertThat(collector.forChannel(source.output()).take(), hasPayload(payload)); // ew hamcrest
    }
}

@@ -682,16 +682,55 @@
 nsion;%0A%0A
+import java.util.concurrent.TimeUnit;%0A%0A
 import s
@@ -774,16 +774,55 @@
 rtThat;%0A
+import static org.hamcrest.core.Is.is;%0A
 import s
@@ -847,27 +847,28 @@
 amework.
-integration
+cloud.stream
 .test.ma
@@ -877,23 +877,28 @@
 her.
-Payload
+MessageQueue
 Matcher.
 hasP
@@ -893,26 +893,31 @@
 Matcher.
-ha
+receive
 sPayload
 ;%0A%0A@Fiel
@@ -908,16 +908,20 @@
 sPayload
+That
 ;%0A%0A@Fiel
@@ -1074,15 +1074,8 @@
 RT)%0A
-public 
 clas
@@ -1108,16 +1108,16 @@
 Tests %7B%0A
+
 %0A    sta
@@ -1571,17 +1571,16 @@
 yThrows(
-%7B
 JsonProc
@@ -1604,37 +1604,8 @@
 lass
-, InterruptedException.class%7D
 )%0A  
@@ -2036,27 +2036,37 @@
 t())
-.take(), ha
+,%0A            receive
 sPayload
 (pay
@@ -2061,16 +2061,23 @@
 sPayload
+That(is
 (payload
@@ -2082,24 +2082,38 @@
 ad))
-; // ew hamcrest
+.within(1, TimeUnit.SECONDS));
 %0A   

Use async matcher in StreamService integration test