code: /*
   * Copyright 2018 Mitsunori Komatsu (komamitsu)
   *
   * 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 org.komamitsu.fluency.integration;
  
  import com.fasterxml.jackson.annotation.JsonProperty;
  import com.fasterxml.jackson.databind.ObjectMapper;
  import org.junit.jupiter.api.Test;
  import org.komamitsu.fluency.Fluency;
  import org.komamitsu.fluency.fluentd.FluencyBuilderForFluentd;
  
  import java.io.IOException;
  import java.net.InetSocketAddress;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.Callable;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  
  import static org.junit.jupiter.api.Assumptions.assumeTrue;
  
  class WithRealFluentd
  {
      private final ObjectMapper objectMapper = new ObjectMapper();
  
      WithRealFluentd.Config getConfig()
              throws IOException
      {
          String conf = System.getenv("WITH_FLUENTD");
          if (conf == null) {
              return null;
          }
  
          return objectMapper.readValue(conf, WithRealFluentd.Config.class);
      }
  
      @Test
      void testWithRealFluentd()
              throws Exception
      {
          WithRealFluentd.Config config = getConfig();
          assumeTrue(config != null);
  
          FluencyBuilderForFluentd builder = new FluencyBuilderForFluentd();
          builder.setSslEnabled(config.sslEnabled);
  
          try (Fluency fluency = builder.build(config.host, config.port)) {
              Map<String, Object> data = new HashMap<>();
              data.put("name", "komamitsu");
              data.put("age", 42);
              data.put("comment", "hello, world");
              ExecutorService executorService = Executors.newCachedThreadPool();
              List<Future<Void>> futures = new ArrayList<>();
              for (int i = 0; i < config.concurrency; i++) {
                  futures.add(executorService.submit(new EmitTask(fluency, config.tag, data, config.requests)));
              }
              for (Future<Void> future : futures) {
                  future.get(config.waitSeconds, TimeUnit.SECONDS);
              }
          }
      }
  
      @Test
      void testWithRealMultipleFluentd()
              throws IOException, InterruptedException, TimeoutException, ExecutionException
      {
          WithRealFluentd.Config config = getConfig();
          assumeTrue(config != null);
          assumeTrue(config.anotherPort != null);
  
          FluencyBuilderForFluentd builder = new FluencyBuilderForFluentd();
          builder.setSslEnabled(config.sslEnabled);
          builder.setAckResponseMode(true);
  
          try (Fluency fluency = builder.build(
                  Arrays.asList(
                          new InetSocketAddress(config.host, config.port),
                          new InetSocketAddress(config.host, config.anotherPort)))) {
  
              Map<String, Object> data = new HashMap<>();
              data.put("name", "komamitsu");
              data.put("age", 42);
              data.put("comment", "hello, world");
              ExecutorService executorService = Executors.newCachedThreadPool();
              List<Future<Void>> futures = new ArrayList<>();
              for (int i = 0; i < config.concurrency; i++) {
                  futures.add(executorService.submit(new EmitTask(fluency, config.tag, data, config.requests)));
              }
              for (Future<Void> future : futures) {
                  future.get(config.waitSeconds, TimeUnit.SECONDS);
              }
          }
      }
  
      @Test
      void testWithRealFluentdWithFileBackup()
              throws ExecutionException, TimeoutException, IOException, InterruptedException
      {
          WithRealFluentd.Config config = getConfig();
          assumeTrue(config != null);
  
          FluencyBuilderForFluentd builder = new FluencyBuilderForFluentd();
          builder.setSslEnabled(config.sslEnabled);
          // Fluency might use a lot of buffer for loaded backup files.
          // So it'd better increase max buffer size
          builder.setMaxBufferSize(512 * 1024 * 1024L);
          builder.setFileBackupDir(System.getProperty("java.io.tmpdir"));
  
          try (Fluency fluency = new FluencyBuilderForFluentd().build(config.host, config.port)) {
              Map<String, Object> data = new HashMap<>();
              data.put("name", "komamitsu");
              data.put("age", 42);
              data.put("comment", "hello, world");
              ExecutorService executorService = Executors.newCachedThreadPool();
              List<Future<Void>> futures = new ArrayList<>();
              for (int i = 0; i < config.concurrency; i++) {
                  futures.add(executorService.submit(new EmitTask(fluency, config.tag, data, config.requests)));
              }
              for (Future<Void> future : futures) {
                  future.get(config.waitSeconds, TimeUnit.SECONDS);
              }
          }
      }
  
      public static class EmitTask
              implements Callable<Void>
      {
          private final Fluency fluency;
          private final String tag;
          private final Map<String, Object> data;
          private final int count;
  
          private EmitTask(Fluency fluency, String tag, Map<String, Object> data, int count)
          {
              this.fluency = fluency;
              this.tag = tag;
              this.data = data;
              this.count = count;
          }
  
          @Override
          public Void call()
          {
              for (int i = 0; i < count; i++) {
                  try {
                      fluency.emit(tag, data);
                  }
                  catch (IOException e) {
                      e.printStackTrace();
                      try {
                          TimeUnit.MILLISECONDS.sleep(500);
                      }
                      catch (InterruptedException e1) {
                          e1.printStackTrace();
                      }
                  }
              }
              return null;
          }
      }
  
      public static class Config
      {
          @JsonProperty("host")
          public final String host;
          @JsonProperty("port")
          public final Integer port;
          @JsonProperty("another_host")
          public final String anotherHost;
          @JsonProperty("another_port")
          public final Integer anotherPort;
          @JsonProperty("tag")
          public final String tag;
          @JsonProperty("requests")
          public final int requests;
          @JsonProperty("concurrency")
          public final int concurrency;
          @JsonProperty("wait_seconds")
          public final int waitSeconds;
          @JsonProperty("ssl_enabled")
          public final boolean sslEnabled;
  
          public Config(
                  @JsonProperty("host")
                          String host,
                  @JsonProperty("port")
                          Integer port,
                  @JsonProperty("another_host")
                          String anotherHost,
                  @JsonProperty("another_port")
                          Integer anotherPort,
                  @JsonProperty("tag")
                          String tag,
                  @JsonProperty("requests")
                          Integer requests,
                  @JsonProperty("concurrency")
                          Integer concurrency,
                  @JsonProperty("wait_seconds")
                          Integer waitSeconds,
                  @JsonProperty("ssl_enabled")
                          Boolean sslEnabled
          )
          {
              this.host = host == null ? "127.0.0.1" : host;
              this.port = port == null ? Integer.valueOf(24224) : port;
  
              this.anotherHost = anotherHost == null ? "127.0.0.1" : anotherHost;
              // Nullable
              this.anotherPort = anotherPort;
  
              this.tag = tag == null ? "foodb.bartbl" : tag;
              this.requests = requests == null ? 1000000 : requests;
              this.concurrency = concurrency == null ? 4 : concurrency;
              this.waitSeconds = waitSeconds == null ? 60 : waitSeconds;
              this.sslEnabled = sslEnabled == null ? false : sslEnabled;
          }
      }
  }
  