diff --git a/stream-applications-integration-tests/pom.xml b/stream-applications-integration-tests/pom.xml index 690e586..8e2bd3a 100644 --- a/stream-applications-integration-tests/pom.xml +++ b/stream-applications-integration-tests/pom.xml @@ -5,7 +5,7 @@ org.springframework.boot spring-boot-starter-parent - 2.3.3.RELEASE + 2.3.4.RELEASE org.springframework.cloud.stream.apps @@ -146,14 +146,27 @@ spring-boot-starter-jdbc test - - org.springframework.boot - spring-boot-starter-webflux - test - + + + + + + + + + + + + + + + + + + org.apache.maven.plugins maven-checkstyle-plugin diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/TikTokBaselineTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/TikTokBaselineTests.java deleted file mode 100644 index 71f397a..0000000 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/TikTokBaselineTests.java +++ /dev/null @@ -1,71 +0,0 @@ -/* - * Copyright 2020-2020 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 - * - * https://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 org.springframework.cloud.stream.apps.integration.test; - -import java.util.regex.Pattern; - -import org.junit.jupiter.api.Test; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.testcontainers.containers.GenericContainer; -import org.testcontainers.junit.jupiter.Container; -import org.testcontainers.junit.jupiter.Testcontainers; -import org.testcontainers.utility.DockerImageName; - -import org.springframework.cloud.stream.app.test.integration.LogMatcher; -import org.springframework.cloud.stream.apps.integration.test.support.KafkaStreamIntegrationTestSupport; - -import static org.springframework.cloud.stream.app.test.integration.FluentMap.fluentMap; - -@Testcontainers -public class TikTokBaselineTests extends KafkaStreamIntegrationTestSupport { - - private static Logger logger = LoggerFactory.getLogger(TikTokBaselineTests.class); - - // "MM/dd/yy HH:mm:ss"; - private final static Pattern pattern = Pattern.compile(".*\\d{2}/\\d{2}/\\d{2}\\s+\\d{2}:\\d{2}:\\d{2}"); - - private final static LogMatcher logMatcher = new LogMatcher(); - - @Container - static GenericContainer timeSource = new GenericContainer( - DockerImageName.parse("springcloudstream/time-source-kafka:3.0.0-SNAPSHOT")) - .withNetwork(kafka.getNetwork()) - .withEnv(fluentMap().withEntry("SPRING_CLOUD_STREAM_BINDINGS_OUTPUT_DESTINATION", "test-topic") - .withEntry("SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS", - kafka.getNetworkAliases().get(0) + ":9092")) - .dependsOn(kafka); - - // @Container - // static GenericContainer logSink = new GenericContainer( - // DockerImageName.parse("springcloudstream/log-sink-kafka:3.0.0-SNAPSHOT")) - // .withNetwork(kafka.getNetwork()) - // .withEnv(fluentMap().withEntry("SPRING_CLOUD_STREAM_BINDINGS_INPUT_DESTINATION", - // "test-topic") - // .withEntry("SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS", - // kafka.getNetworkAliases().get(0) + ":9092") - // .withEntry("SPRING_CLOUD_STREAM_BINDINGS_INPUT_GROUP", "TikTok")) - // .withLogConsumer(logMatcher) - // // .withLogConsumer(appLog("log-sink")) - // .dependsOn(kafka); - - @Test - void tiktok() { - // await().atMost(Duration.ofMinutes(2)).until(verifyOutputMessage((Message m) -> - // pattern.matcher(m.getPayload()).matches())); - } -} diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/TikTokTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/TikTokTests.java deleted file mode 100644 index ce4c362..0000000 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/TikTokTests.java +++ /dev/null @@ -1,51 +0,0 @@ -/* - * Copyright 2020-2020 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 - * - * https://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 org.springframework.cloud.stream.apps.integration.test; - -import java.time.Duration; -import java.util.regex.Pattern; - -import org.junit.jupiter.api.Test; -import org.testcontainers.junit.jupiter.Container; - -import org.springframework.cloud.stream.app.test.integration.LogMatcher; -import org.springframework.cloud.stream.app.test.integration.StreamApps; -import org.springframework.cloud.stream.apps.integration.test.support.KafkaStreamIntegrationTestSupport; - -import static org.awaitility.Awaitility.await; -import static org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps.kafkaStreamApps; - -public class TikTokTests extends KafkaStreamIntegrationTestSupport { - // "MM/dd/yy HH:mm:ss"; - private final static Pattern DATE_PATTERN = Pattern.compile(".*\\d{2}/\\d{2}/\\d{2}\\s+\\d{2}:\\d{2}:\\d{2}"); - - private final static LogMatcher logMatcher = new LogMatcher(); -// -// @Container -// static StreamApps streamApps = kafkaStreamApps("tikTok", kafka) -// .withSourceContainer(defaultKafkaContainerFor("time-source")) -// .withSinkContainer(defaultKafkaContainerFor("log-sink") -// .withLogConsumer(logMatcher)) -// .build(); -// -// @Test -// void tiktok() { -// await().atMost(Duration.ofMinutes(2)).until(logMatcher.verifies(log -> log.contains("Started LogSink"))); -// await().atMost(Duration.ofSeconds(30)) -// .until(logMatcher.verifies(log -> log.matchesRegex(DATE_PATTERN.pattern()))); -// } -} diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/processor/HttpRequestProcessorTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/processor/HttpRequestProcessorTests.java index c6a1afb..df1cfa3 100644 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/processor/HttpRequestProcessorTests.java +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/processor/HttpRequestProcessorTests.java @@ -16,31 +16,12 @@ package org.springframework.cloud.stream.apps.integration.test.processor; -import java.net.InetAddress; -import java.time.Duration; - -import okhttp3.mockwebserver.Dispatcher; -import okhttp3.mockwebserver.MockResponse; import okhttp3.mockwebserver.MockWebServer; -import okhttp3.mockwebserver.RecordedRequest; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Test; -import org.testcontainers.junit.jupiter.Container; -import reactor.core.publisher.Mono; import org.springframework.cloud.stream.app.test.integration.LogMatcher; -import org.springframework.cloud.stream.app.test.integration.StreamApps; import org.springframework.cloud.stream.apps.integration.test.support.KafkaStreamIntegrationTestSupport; -import org.springframework.http.HttpHeaders; -import org.springframework.http.HttpStatus; -import org.springframework.http.MediaType; -import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.WebClient; -import static org.assertj.core.api.Assertions.assertThat; -import static org.awaitility.Awaitility.await; -import static org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps.kafkaStreamApps; - public class HttpRequestProcessorTests extends KafkaStreamIntegrationTestSupport { private static MockWebServer server = new MockWebServer(); @@ -52,45 +33,46 @@ public class HttpRequestProcessorTests extends KafkaStreamIntegrationTestSupport private static int sourcePort = findAvailablePort(); -// @Container -// private static final StreamApps streamApps = kafkaStreamApps( -// HttpRequestProcessorTests.class.getSimpleName(), kafka) -// .withSourceContainer(httpSource(sourcePort)) -// .withProcessorContainer(defaultKafkaProcessorContainerFor("http-request-processor") -// .withEnv("HTTP_REQUEST_URL_EXPRESSION", -// "'http://" + localHostAddress() + ":" + serverPort + "'") -// .withEnv("HTTP_REQUEST_HTTP_METHOD_EXPRESSION", "'POST'")) -// .withSinkContainer( -// defaultKafkaContainerFor("log-sink").withLogConsumer(logMatcher)) -// .build(); -// -// @BeforeAll -// static void startServer() throws Exception { -// server.start(InetAddress.getLocalHost(), serverPort); -// } -// -// @Test -// void get() { -// server.setDispatcher(new Dispatcher() { -// @Override -// public MockResponse dispatch(RecordedRequest recordedRequest) { -// return new MockResponse().setHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) -// .setBody("{\"response\":\"" + recordedRequest.getBody().readUtf8() + "\"}") -// .setResponseCode(HttpStatus.OK.value()); -// } -// }); -// -// await().atMost(Duration.ofSeconds(30)) -// .until(logMatcher.verifies(log -> log.when(() -> { -// ClientResponse response = webClient -// .post() -// .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(sourcePort)) -// .contentType(MediaType.TEXT_PLAIN) -// .body(Mono.just("ping"), String.class) -// .exchange() -// .block(); -// assertThat(response.statusCode().is2xxSuccessful()).isTrue(); -// }).matchesRegex(".*\\{\"response\":\"ping\"\\}"))); -// } + // @Container + // private static final StreamApps streamApps = kafkaStreamApps( + // HttpRequestProcessorTests.class.getSimpleName(), kafka) + // .withSourceContainer(httpSource(sourcePort)) + // .withProcessorContainer(defaultKafkaProcessorContainerFor("http-request-processor") + // .withEnv("HTTP_REQUEST_URL_EXPRESSION", + // "'http://" + localHostAddress() + ":" + serverPort + "'") + // .withEnv("HTTP_REQUEST_HTTP_METHOD_EXPRESSION", "'POST'")) + // .withSinkContainer( + // defaultKafkaContainerFor("log-sink").withLogConsumer(logMatcher)) + // .build(); + // + // @BeforeAll + // static void startServer() throws Exception { + // server.start(InetAddress.getLocalHost(), serverPort); + // } + // + // @Test + // void get() { + // server.setDispatcher(new Dispatcher() { + // @Override + // public MockResponse dispatch(RecordedRequest recordedRequest) { + // return new MockResponse().setHeader(HttpHeaders.CONTENT_TYPE, + // MediaType.APPLICATION_JSON_VALUE) + // .setBody("{\"response\":\"" + recordedRequest.getBody().readUtf8() + "\"}") + // .setResponseCode(HttpStatus.OK.value()); + // } + // }); + // + // await().atMost(Duration.ofSeconds(30)) + // .until(logMatcher.verifies(log -> log.when(() -> { + // ClientResponse response = webClient + // .post() + // .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(sourcePort)) + // .contentType(MediaType.TEXT_PLAIN) + // .body(Mono.just("ping"), String.class) + // .exchange() + // .block(); + // assertThat(response.statusCode().is2xxSuccessful()).isTrue(); + // }).matchesRegex(".*\\{\"response\":\"ping\"\\}"))); + // } } diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/JdbcSinkTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/JdbcSinkTests.java index 2a5b5ec..cbb7118 100644 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/JdbcSinkTests.java +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/JdbcSinkTests.java @@ -16,29 +16,10 @@ package org.springframework.cloud.stream.apps.integration.test.sink; -import java.time.Duration; - -import com.zaxxer.hikari.HikariDataSource; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Test; -import org.testcontainers.containers.BindMode; -import org.testcontainers.containers.MySQLContainer; -import org.testcontainers.containers.wait.strategy.Wait; -import org.testcontainers.junit.jupiter.Container; -import org.testcontainers.utility.DockerImageName; -import reactor.core.publisher.Mono; - -import org.springframework.cloud.stream.app.test.integration.StreamApps; import org.springframework.cloud.stream.apps.integration.test.support.KafkaStreamIntegrationTestSupport; -import org.springframework.http.MediaType; import org.springframework.jdbc.core.JdbcTemplate; -import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.WebClient; -import static org.assertj.core.api.Assertions.assertThat; -import static org.awaitility.Awaitility.await; -import static org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps.kafkaStreamApps; - public class JdbcSinkTests extends KafkaStreamIntegrationTestSupport { private static int serverPort = findAvailablePort(); @@ -47,60 +28,63 @@ public class JdbcSinkTests extends KafkaStreamIntegrationTestSupport { private static WebClient webClient = WebClient.builder().build(); -// @Container -// private static MySQLContainer mySQL = new MySQLContainer<>(DockerImageName.parse("mysql:5.7")) -// .withUsername("test") -// .withPassword("secret") -// .withExposedPorts(3306) -// .withNetwork(kafka.getNetwork()) -// .withClasspathResourceMapping("init.sql", "/init.sql", BindMode.READ_ONLY) -// .withCommand("--init-file", "/init.sql"); -// -// @Container -// private static StreamApps streamApps = kafkaStreamApps(JdbcSinkTests.class.getSimpleName(), kafka) -// .withSourceContainer(defaultKafkaContainerFor("http-source") -// .withEnv("SERVER_PORT", String.valueOf(serverPort)) -// .withExposedPorts(serverPort) -// .waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(2)))) -// .withSinkContainer(defaultKafkaContainerFor("jdbc-sink") -// .withEnv("JDBC_CONSUMER_COLUMNS", "name,city:address.city,street:address.street") -// .withEnv("JDBC_CONSUMER_TABLE_NAME", "People") -// .withEnv("SPRING_DATASOURCE_USERNAME", "test") -// .withEnv("SPRING_DATASOURCE_PASSWORD", "secret") -// .withEnv("SPRING_DATASOURCE_DRIVER_CLASS_NAME", "org.mariadb.jdbc.Driver") -// .withEnv("SPRING_DATASOURCE_URL", -// "jdbc:mysql://" + mySQL.getNetworkAliases().get(0) + ":3306/test")) -// .build(); -// -// @BeforeAll -// static void startStreamApps() { -// HikariDataSource dataSource = new HikariDataSource(); -// dataSource.setDriverClassName("org.mariadb.jdbc.Driver"); -// dataSource.setUsername(mySQL.getUsername()); -// dataSource.setPassword(mySQL.getPassword()); -// dataSource.setJdbcUrl("jdbc:mysql://localhost:" + mySQL.getMappedPort(3306) + "/test"); -// jdbcTemplate = new JdbcTemplate(dataSource); -// jdbcTemplate.execute("DELETE FROM People"); -// } -// -// @Test -// void postData() { -// String json = "{\"name\":\"My Name\",\"address\":{ \"city\": \"Big City\", \"street\":\"Narrow Alley\"}}"; -// ClientResponse response = webClient -// .post() -// .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(serverPort)) -// .contentType(MediaType.APPLICATION_JSON) -// .body(Mono.just(json), String.class) -// .exchange() -// .block(); -// assertThat(response.statusCode().is2xxSuccessful()).isTrue(); -// -// await().atMost(Duration.ofSeconds(30)) -// .untilAsserted( -// () -> assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) from People", -// Integer.class)) -// .isOne()); -// assertThat(jdbcTemplate.queryForObject("SELECT name from People", -// String.class)).isEqualTo("My Name"); -// } + // @Container + // private static MySQLContainer mySQL = new + // MySQLContainer<>(DockerImageName.parse("mysql:5.7")) + // .withUsername("test") + // .withPassword("secret") + // .withExposedPorts(3306) + // .withNetwork(kafka.getNetwork()) + // .withClasspathResourceMapping("init.sql", "/init.sql", BindMode.READ_ONLY) + // .withCommand("--init-file", "/init.sql"); + // + // @Container + // private static StreamApps streamApps = + // kafkaStreamApps(JdbcSinkTests.class.getSimpleName(), kafka) + // .withSourceContainer(defaultKafkaContainerFor("http-source") + // .withEnv("SERVER_PORT", String.valueOf(serverPort)) + // .withExposedPorts(serverPort) + // .waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(2)))) + // .withSinkContainer(defaultKafkaContainerFor("jdbc-sink") + // .withEnv("JDBC_CONSUMER_COLUMNS", "name,city:address.city,street:address.street") + // .withEnv("JDBC_CONSUMER_TABLE_NAME", "People") + // .withEnv("SPRING_DATASOURCE_USERNAME", "test") + // .withEnv("SPRING_DATASOURCE_PASSWORD", "secret") + // .withEnv("SPRING_DATASOURCE_DRIVER_CLASS_NAME", "org.mariadb.jdbc.Driver") + // .withEnv("SPRING_DATASOURCE_URL", + // "jdbc:mysql://" + mySQL.getNetworkAliases().get(0) + ":3306/test")) + // .build(); + // + // @BeforeAll + // static void startStreamApps() { + // HikariDataSource dataSource = new HikariDataSource(); + // dataSource.setDriverClassName("org.mariadb.jdbc.Driver"); + // dataSource.setUsername(mySQL.getUsername()); + // dataSource.setPassword(mySQL.getPassword()); + // dataSource.setJdbcUrl("jdbc:mysql://localhost:" + mySQL.getMappedPort(3306) + "/test"); + // jdbcTemplate = new JdbcTemplate(dataSource); + // jdbcTemplate.execute("DELETE FROM People"); + // } + // + // @Test + // void postData() { + // String json = "{\"name\":\"My Name\",\"address\":{ \"city\": \"Big City\", + // \"street\":\"Narrow Alley\"}}"; + // ClientResponse response = webClient + // .post() + // .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(serverPort)) + // .contentType(MediaType.APPLICATION_JSON) + // .body(Mono.just(json), String.class) + // .exchange() + // .block(); + // assertThat(response.statusCode().is2xxSuccessful()).isTrue(); + // + // await().atMost(Duration.ofSeconds(30)) + // .untilAsserted( + // () -> assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) from People", + // Integer.class)) + // .isOne()); + // assertThat(jdbcTemplate.queryForObject("SELECT name from People", + // String.class)).isEqualTo("My Name"); + // } } diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/MongoDBSinkTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/MongoDBSinkTests.java index 0fbf379..7f6e745 100644 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/MongoDBSinkTests.java +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/MongoDBSinkTests.java @@ -16,73 +16,57 @@ package org.springframework.cloud.stream.apps.integration.test.sink; -import java.time.Duration; -import java.util.List; - -import org.bson.Document; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Test; -import org.testcontainers.containers.MongoDBContainer; -import org.testcontainers.junit.jupiter.Container; -import org.testcontainers.utility.DockerImageName; -import reactor.core.publisher.Mono; - -import org.springframework.cloud.stream.app.test.integration.StreamApps; import org.springframework.cloud.stream.apps.integration.test.support.KafkaStreamIntegrationTestSupport; -import org.springframework.data.mongodb.MongoDatabaseFactory; -import org.springframework.data.mongodb.core.MongoTemplate; -import org.springframework.data.mongodb.core.SimpleMongoClientDatabaseFactory; -import org.springframework.http.MediaType; -import org.springframework.web.reactive.function.client.ClientResponse; -import org.springframework.web.reactive.function.client.WebClient; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps.kafkaStreamApps; public class MongoDBSinkTests extends KafkaStreamIntegrationTestSupport { -// -// private static int serverPort = findAvailablePort(); -// -// private static MongoTemplate mongoTemplate; -// -// private static WebClient webClient = WebClient.builder().build(); -// -// @Container -// private static MongoDBContainer mongoDBContainer = new MongoDBContainer(DockerImageName.parse("mongo:4.0.10")) -// .withExposedPorts(27017) -// .withStartupTimeout(Duration.ofMinutes(2)); -// -// private static String mongoConnectionString() { -// return String.format("mongodb://%s:%s/%s", localHostAddress(), mongoDBContainer.getMappedPort(27017), "test"); -// } -// -// @Container -// private StreamApps streamApps = kafkaStreamApps(MongoDBSinkTests.class.getSimpleName(), kafka) -// .withSourceContainer(httpSource(serverPort)) -// .withSinkContainer(defaultKafkaContainerFor("mongodb-sink") -// .withEnv("MONGO_DB_CONSUMER_COLLECTION", "test") -// .withEnv("SPRING_DATA_MONGODB_URL", mongoConnectionString())) -// .build(); -// -// @BeforeAll -// static void buildMongoTemplate() { -// MongoDatabaseFactory mongoDatabaseFactory = new SimpleMongoClientDatabaseFactory( -// mongoConnectionString()); -// mongoTemplate = new MongoTemplate(mongoDatabaseFactory); -// } -// -// @Test -// void postData() { -// String json = "{\"name\":\"My Name\",\"address\":{ \"city\": \"Big City\", \"street\": \"Narrow Alley\"}}"; -// ClientResponse response = webClient -// .post() -// .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(serverPort)) -// .contentType(MediaType.APPLICATION_JSON) -// .body(Mono.just(json), String.class) -// .exchange() -// .block(Duration.ofSeconds(30)); -// assertThat(response.statusCode().is2xxSuccessful()).isTrue(); -// List docs = mongoTemplate.findAll(Document.class, "test"); -// assertThat(docs).allMatch(document -> document.get("name", String.class).equals("My Name")); -// } + // + // private static int serverPort = findAvailablePort(); + // + // private static MongoTemplate mongoTemplate; + // + // private static WebClient webClient = WebClient.builder().build(); + // + // @Container + // private static MongoDBContainer mongoDBContainer = new + // MongoDBContainer(DockerImageName.parse("mongo:4.0.10")) + // .withExposedPorts(27017) + // .withStartupTimeout(Duration.ofMinutes(2)); + // + // private static String mongoConnectionString() { + // return String.format("mongodb://%s:%s/%s", localHostAddress(), + // mongoDBContainer.getMappedPort(27017), "test"); + // } + // + // @Container + // private StreamApps streamApps = kafkaStreamApps(MongoDBSinkTests.class.getSimpleName(), + // kafka) + // .withSourceContainer(httpSource(serverPort)) + // .withSinkContainer(defaultKafkaContainerFor("mongodb-sink") + // .withEnv("MONGO_DB_CONSUMER_COLLECTION", "test") + // .withEnv("SPRING_DATA_MONGODB_URL", mongoConnectionString())) + // .build(); + // + // @BeforeAll + // static void buildMongoTemplate() { + // MongoDatabaseFactory mongoDatabaseFactory = new SimpleMongoClientDatabaseFactory( + // mongoConnectionString()); + // mongoTemplate = new MongoTemplate(mongoDatabaseFactory); + // } + // + // @Test + // void postData() { + // String json = "{\"name\":\"My Name\",\"address\":{ \"city\": \"Big City\", \"street\": + // \"Narrow Alley\"}}"; + // ClientResponse response = webClient + // .post() + // .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(serverPort)) + // .contentType(MediaType.APPLICATION_JSON) + // .body(Mono.just(json), String.class) + // .exchange() + // .block(Duration.ofSeconds(30)); + // assertThat(response.statusCode().is2xxSuccessful()).isTrue(); + // List docs = mongoTemplate.findAll(Document.class, "test"); + // assertThat(docs).allMatch(document -> document.get("name", String.class).equals("My + // Name")); + // } } diff --git a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/TcpSinkTests.java b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/TcpSinkTests.java index 92b8b0a..bc143ba 100644 --- a/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/TcpSinkTests.java +++ b/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/apps/integration/test/sink/TcpSinkTests.java @@ -16,77 +16,58 @@ package org.springframework.cloud.stream.apps.integration.test.sink; -import java.io.BufferedReader; -import java.io.IOException; -import java.io.InputStreamReader; -import java.net.InetAddress; -import java.net.ServerSocket; -import java.net.Socket; -import java.time.Duration; -import java.util.concurrent.atomic.AtomicBoolean; - -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Test; -import org.testcontainers.junit.jupiter.Container; -import reactor.core.publisher.Mono; - -import org.springframework.cloud.stream.app.test.integration.StreamApps; import org.springframework.cloud.stream.apps.integration.test.support.KafkaStreamIntegrationTestSupport; -import org.springframework.http.MediaType; -import org.springframework.web.reactive.function.client.ClientResponse; -import org.springframework.web.reactive.function.client.WebClient; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.awaitility.Awaitility.await; -import static org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamApps.kafkaStreamApps; public class TcpSinkTests extends KafkaStreamIntegrationTestSupport { -// private static final int port = findAvailablePort(); -// -// private static final int tcpPort = findAvailablePort(); -// -// private static Socket socket; -// -// private static final AtomicBoolean socketReady = new AtomicBoolean(); -// -// private static WebClient webClient = WebClient.builder().build(); -// -// @Container -// private static StreamApps streamApps = kafkaStreamApps(TcpSinkTests.class.getSimpleName(), kafka) -// .withSourceContainer(httpSource(port)) -// .withSinkContainer(defaultKafkaContainerFor("tcp-sink") -// .withEnv("TCP_CONSUMER_HOST", localHostAddress()) -// .withEnv("TCP_PORT", String.valueOf(tcpPort)) -// .withEnv("TCP_CONSUMER_ENCODER", "CRLF")) -// .build(); -// -// @BeforeAll -// static void startTcpServer() { -// new Thread(() -> { -// try { -// socket = new ServerSocket(tcpPort, 50, InetAddress.getLocalHost()).accept(); -// socketReady.set(true); -// } -// catch (IOException exception) { -// exception.printStackTrace(); -// } -// }).start(); -// } -// -// @Test -// void postData() throws IOException { -// String text = "Hello, world!"; -// ClientResponse response = webClient -// .post() -// .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(port)) -// .contentType(MediaType.TEXT_PLAIN) -// .body(Mono.just(text), String.class) -// .exchange() -// .block(); -// assertThat(response.statusCode().is2xxSuccessful()).isTrue(); -// await().atMost(Duration.ofSeconds(10)).untilTrue(socketReady); -// BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream())); -// await().atMost(Duration.ofSeconds(10)).until(() -> reader.readLine().equals("Hello, world!")); -// } + // private static final int port = findAvailablePort(); + // + // private static final int tcpPort = findAvailablePort(); + // + // private static Socket socket; + // + // private static final AtomicBoolean socketReady = new AtomicBoolean(); + // + // private static WebClient webClient = WebClient.builder().build(); + // + // @Container + // private static StreamApps streamApps = + // kafkaStreamApps(TcpSinkTests.class.getSimpleName(), kafka) + // .withSourceContainer(httpSource(port)) + // .withSinkContainer(defaultKafkaContainerFor("tcp-sink") + // .withEnv("TCP_CONSUMER_HOST", localHostAddress()) + // .withEnv("TCP_PORT", String.valueOf(tcpPort)) + // .withEnv("TCP_CONSUMER_ENCODER", "CRLF")) + // .build(); + // + // @BeforeAll + // static void startTcpServer() { + // new Thread(() -> { + // try { + // socket = new ServerSocket(tcpPort, 50, InetAddress.getLocalHost()).accept(); + // socketReady.set(true); + // } + // catch (IOException exception) { + // exception.printStackTrace(); + // } + // }).start(); + // } + // + // @Test + // void postData() throws IOException { + // String text = "Hello, world!"; + // ClientResponse response = webClient + // .post() + // .uri("http://localhost:" + streamApps.sourceContainer().getMappedPort(port)) + // .contentType(MediaType.TEXT_PLAIN) + // .body(Mono.just(text), String.class) + // .exchange() + // .block(); + // assertThat(response.statusCode().is2xxSuccessful()).isTrue(); + // await().atMost(Duration.ofSeconds(10)).untilTrue(socketReady); + // BufferedReader reader = new BufferedReader(new + // InputStreamReader(socket.getInputStream())); + // await().atMost(Duration.ofSeconds(10)).until(() -> reader.readLine().equals("Hello, + // world!")); + // } }