SplunkIO { public static Write write(String url, String token) { checkNotNull(url, "url is required."); checkNotNull(token, "token is required."); return write(StaticValueProvider.of(url), StaticValueProvider.of(token)); } private  SplunkIO(); static Write write(String url, String token); static Write write(ValueProvider<String> url, ValueProvider<String> token);  }
@Test @Category(NeedsRunner.class) public void successfulSplunkIOMultiBatchNoParallelismTest() { mockServerListening(200); int testPort = mockServerRule.getPort(); String url = Joiner.on(':').join("http: String token = "test-token"; List<SplunkEvent> testEvents = ImmutableList.of( SplunkEvent.newBuilder() .withEvent("test-event-1") .withHost("test-host-1") .withIndex("test-index-1") .withSource("test-source-1") .withSourceType("test-source-type-1") .withTime(12345L) .create(), SplunkEvent.newBuilder() .withEvent("test-event-2") .withHost("test-host-2") .withIndex("test-index-2") .withSource("test-source-2") .withSourceType("test-source-type-2") .withTime(12345L) .create()); PCollection<SplunkWriteError> actual = pipeline .apply("Create Input data", Create.of(testEvents)) .apply( "SplunkIO", SplunkIO.write(url, token).withParallelism(1).withBatchCount(testEvents.size())); PAssert.that(actual).empty(); pipeline.run(); mockServerClient.verify(HttpRequest.request(EXPECTED_PATH), VerificationTimes.once()); }
@Test @Category(NeedsRunner.class) public void successfulSplunkIOMultiBatchParallelismTest() { mockServerListening(200); int testPort = mockServerRule.getPort(); int testParallelism = 2; String url = Joiner.on(':').join("http: String token = "test-token"; List<SplunkEvent> testEvents = ImmutableList.of( SplunkEvent.newBuilder() .withEvent("test-event-1") .withHost("test-host-1") .withIndex("test-index-1") .withSource("test-source-1") .withSourceType("test-source-type-1") .withTime(12345L) .create(), SplunkEvent.newBuilder() .withEvent("test-event-2") .withHost("test-host-2") .withIndex("test-index-2") .withSource("test-source-2") .withSourceType("test-source-type-2") .withTime(12345L) .create()); PCollection<SplunkWriteError> actual = pipeline .apply("Create Input data", Create.of(testEvents)) .apply( "SplunkIO", SplunkIO.write(url, token) .withParallelism(testParallelism) .withBatchCount(testEvents.size())); PAssert.that(actual).empty(); pipeline.run(); mockServerClient.verify( HttpRequest.request(EXPECTED_PATH), VerificationTimes.atMost(testParallelism)); }
@Test @Category(NeedsRunner.class) public void successfulSplunkIOSingleBatchParallelismTest() { mockServerListening(200); int testPort = mockServerRule.getPort(); int testParallelism = 2; String url = Joiner.on(':').join("http: String token = "test-token"; List<SplunkEvent> testEvents = ImmutableList.of( SplunkEvent.newBuilder() .withEvent("test-event-1") .withHost("test-host-1") .withIndex("test-index-1") .withSource("test-source-1") .withSourceType("test-source-type-1") .withTime(12345L) .create(), SplunkEvent.newBuilder() .withEvent("test-event-2") .withHost("test-host-2") .withIndex("test-index-2") .withSource("test-source-2") .withSourceType("test-source-type-2") .withTime(12345L) .create()); PCollection<SplunkWriteError> actual = pipeline .apply("Create Input data", Create.of(testEvents)) .apply( "SplunkIO", SplunkIO.write(url, token).withParallelism(testParallelism).withBatchCount(1)); PAssert.that(actual).empty(); pipeline.run(); mockServerClient.verify( HttpRequest.request(EXPECTED_PATH), VerificationTimes.exactly(testEvents.size())); }