code: /*
   * Copyright 2019 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.fluentd;
  
  import org.komamitsu.fluency.Fluency;
  import org.komamitsu.fluency.fluentd.ingester.FluentdIngester;
  import org.komamitsu.fluency.fluentd.ingester.sender.FluentdSender;
  import org.komamitsu.fluency.fluentd.ingester.sender.MultiSender;
  import org.komamitsu.fluency.fluentd.ingester.sender.RetryableSender;
  import org.komamitsu.fluency.fluentd.ingester.sender.SSLSender;
  import org.komamitsu.fluency.fluentd.ingester.sender.TCPSender;
  import org.komamitsu.fluency.fluentd.ingester.sender.failuredetect.FailureDetector;
  import org.komamitsu.fluency.fluentd.ingester.sender.failuredetect.PhiAccrualFailureDetectStrategy;
  import org.komamitsu.fluency.fluentd.ingester.sender.heartbeat.SSLHeartbeater;
  import org.komamitsu.fluency.fluentd.ingester.sender.heartbeat.TCPHeartbeater;
  import org.komamitsu.fluency.fluentd.ingester.sender.retry.ExponentialBackOffRetryStrategy;
  import org.komamitsu.fluency.fluentd.recordformat.FluentdRecordFormatter;
  import org.komamitsu.fluency.ingester.Ingester;
  
  import java.net.InetSocketAddress;
  import javax.net.ssl.SSLSocketFactory;
  import java.util.ArrayList;
  import java.util.List;
  
  public class FluencyBuilderForFluentd
          extends org.komamitsu.fluency.FluencyBuilder
  {
      private Integer senderMaxRetryCount;
      private Integer senderBaseRetryIntervalMillis;
      private Integer senderMaxRetryIntervalMillis;
      private boolean ackResponseMode;
      private boolean sslEnabled;
      private SSLSocketFactory sslSocketFactory;
      protected Integer connectionTimeoutMilli;
      protected Integer readTimeoutMilli;
      protected FluentdRecordFormatter recordFormatter = new FluentdRecordFormatter();
  
      public Integer getSenderMaxRetryCount()
      {
          return senderMaxRetryCount;
      }
  
      public Integer getSenderBaseRetryIntervalMillis()
      {
          return senderBaseRetryIntervalMillis;
      }
  
      public Integer getSenderMaxRetryIntervalMillis()
      {
          return senderMaxRetryIntervalMillis;
      }
  
      public void setSenderBaseRetryIntervalMillis(Integer senderBaseRetryIntervalMillis)
      {
          this.senderBaseRetryIntervalMillis = senderBaseRetryIntervalMillis;
      }
  
      public void setSenderMaxRetryIntervalMillis(Integer senderMaxRetryIntervalMillis)
      {
          this.senderMaxRetryIntervalMillis = senderMaxRetryIntervalMillis;
      }
  
      public void setSenderMaxRetryCount(Integer senderMaxRetryCount)
      {
          this.senderMaxRetryCount = senderMaxRetryCount;
      }
  
      public boolean isAckResponseMode()
      {
          return ackResponseMode;
      }
  
      public void setAckResponseMode(boolean ackResponseMode)
      {
          this.ackResponseMode = ackResponseMode;
      }
  
      public boolean isSslEnabled()
      {
          return sslEnabled;
      }
  
      public void setSslEnabled(boolean sslEnabled)
      {
          this.sslEnabled = sslEnabled;
      }
  
      public void setSslSocketFactory(SSLSocketFactory sslSocketFactory)
      {
          this.sslSocketFactory = sslSocketFactory;
      }
  
      public Integer getConnectionTimeoutMilli()
      {
          return connectionTimeoutMilli;
      }
  
      public void setConnectionTimeoutMilli(Integer connectionTimeoutMilli)
      {
          this.connectionTimeoutMilli = connectionTimeoutMilli;
      }
  
      public Integer getReadTimeoutMilli()
      {
          return readTimeoutMilli;
      }
  
      public void setReadTimeoutMilli(Integer readTimeoutMilli)
      {
          this.readTimeoutMilli = readTimeoutMilli;
      }
  
      public FluentdRecordFormatter getRecordFormatter()
      {
          return recordFormatter;
      }
  
      public void setRecordFormatter(FluentdRecordFormatter recordFormatter)
      {
          this.recordFormatter = recordFormatter;
      }
  
      public Fluency build(String host, int port)
      {
          return buildFromIngester(
                  recordFormatter,
                  buildIngester(createBaseSender(host, port)));
      }
  
      public Fluency build(int port)
      {
          return buildFromIngester(
                  recordFormatter,
                  buildIngester(createBaseSender(null, port)));
      }
  
      public Fluency build()
      {
          return buildFromIngester(
                  recordFormatter,
                  buildIngester(createBaseSender(null, null)));
      }
  
      public Fluency build(List<InetSocketAddress> servers)
      {
          List<FluentdSender> senders = new ArrayList<>();
          for (InetSocketAddress server : servers) {
              senders.add(createBaseSender(server.getHostName(), server.getPort(), true));
          }
          return buildFromIngester(
                  recordFormatter,
                  buildIngester(new MultiSender(senders)));
      }
  
      private FluentdSender createBaseSender(String host, Integer port)
      {
          return createBaseSender(host, port, false);
      }
  
      private FluentdSender createBaseSender(String host, Integer port, boolean withHeartBeater)
      {
          if (withHeartBeater && port == null) {
              throw new IllegalArgumentException("`port` should be specified when using heartbeat");
          }
  
          if (isSslEnabled()) {
              SSLSender.Config senderConfig = new SSLSender.Config();
              FailureDetector failureDetector = null;
              if (host != null) {
                  senderConfig.setHost(host);
              }
              if (port != null) {
                  senderConfig.setPort(port);
              }
              if (sslSocketFactory != null) {
                  senderConfig.setSslSocketFactory(sslSocketFactory);
              }
              if (withHeartBeater) {
                  SSLHeartbeater.Config hbConfig = new SSLHeartbeater.Config();
                  hbConfig.setHost(host);
                  hbConfig.setPort(port);
  
                  if (sslSocketFactory != null) {
                      hbConfig.setSslSocketFactory(sslSocketFactory);
                  }
  
                  SSLHeartbeater heartbeater = new SSLHeartbeater(hbConfig);
                  failureDetector = new FailureDetector(new PhiAccrualFailureDetectStrategy(), heartbeater);
              }
              if (connectionTimeoutMilli != null) {
                  senderConfig.setConnectionTimeoutMilli(connectionTimeoutMilli);
              }
              if (readTimeoutMilli != null) {
                  senderConfig.setReadTimeoutMilli(readTimeoutMilli);
              }
              return new SSLSender(senderConfig, failureDetector);
          }
          else {
              TCPSender.Config senderConfig = new TCPSender.Config();
              FailureDetector failureDetector = null;
              if (host != null) {
                  senderConfig.setHost(host);
              }
              if (port != null) {
                  senderConfig.setPort(port);
              }
              if (withHeartBeater) {
                  TCPHeartbeater.Config hbConfig = new TCPHeartbeater.Config();
                  hbConfig.setHost(host);
                  hbConfig.setPort(port);
                  TCPHeartbeater heartbeater = new TCPHeartbeater(hbConfig);
                  failureDetector = new FailureDetector(new PhiAccrualFailureDetectStrategy(), heartbeater);
              }
              if (connectionTimeoutMilli != null) {
                  senderConfig.setConnectionTimeoutMilli(connectionTimeoutMilli);
              }
              if (readTimeoutMilli != null) {
                  senderConfig.setReadTimeoutMilli(readTimeoutMilli);
              }
              return new TCPSender(senderConfig, failureDetector);
          }
      }
  
      @Override
      public String toString()
      {
          return "FluencyBuilder{" +
                  "senderMaxRetryCount=" + senderMaxRetryCount +
                  ", ackResponseMode=" + ackResponseMode +
                  ", sslEnabled=" + sslEnabled +
                  "} " + super.toString();
      }
  
      protected Ingester buildIngester(FluentdSender baseSender)
      {
          ExponentialBackOffRetryStrategy.Config retryStrategyConfig =
                  new ExponentialBackOffRetryStrategy.Config();
  
          if (getSenderMaxRetryCount() != null) {
              retryStrategyConfig.setMaxRetryCount(getSenderMaxRetryCount());
          }
  
          if (getSenderBaseRetryIntervalMillis() != null) {
              retryStrategyConfig.setBaseIntervalMillis(getSenderBaseRetryIntervalMillis());
          }
  
          if (getSenderMaxRetryIntervalMillis() != null) {
              retryStrategyConfig.setMaxIntervalMillis(getSenderMaxRetryIntervalMillis());
          }
  
          RetryableSender.Config senderConfig = new RetryableSender.Config();
  
          if (getErrorHandler() != null) {
              senderConfig.setErrorHandler(getErrorHandler());
          }
  
          RetryableSender retryableSender = new RetryableSender(senderConfig, baseSender,
                  new ExponentialBackOffRetryStrategy(retryStrategyConfig));
  
          FluentdIngester.Config ingesterConfig = new FluentdIngester.Config();
          ingesterConfig.setAckResponseMode(isAckResponseMode());
  
          return new FluentdIngester(ingesterConfig, retryableSender);
      }
  }
  