package com.dbnt.faisp.config.tomcat;

import org.apache.catalina.Context;
import org.apache.catalina.Engine;
import org.apache.catalina.ha.session.ClusterSessionListener;
import org.apache.catalina.ha.session.DeltaManager;
import org.apache.catalina.ha.session.JvmRouteBinderValve;
import org.apache.catalina.ha.tcp.ReplicationValve;
import org.apache.catalina.ha.tcp.SimpleTcpCluster;
import org.apache.catalina.tribes.group.GroupChannel;
import org.apache.catalina.tribes.group.interceptors.MessageDispatchInterceptor;
import org.apache.catalina.tribes.group.interceptors.StaticMembershipInterceptor;
import org.apache.catalina.tribes.group.interceptors.TcpFailureDetector;
import org.apache.catalina.tribes.group.interceptors.TcpPingInterceptor;
import org.apache.catalina.tribes.membership.McastService;
import org.apache.catalina.tribes.membership.StaticMember;
import org.apache.catalina.tribes.transport.ReplicationTransmitter;
import org.apache.catalina.tribes.transport.nio.NioReceiver;
import org.apache.catalina.tribes.transport.nio.PooledParallelSender;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.web.embedded.tomcat.TomcatContextCustomizer;
import org.springframework.stereotype.Component;

@Component
public class TomcatClusterContextCustomizer implements TomcatContextCustomizer {

  @Value("${tomcat.cluster.static-member-host:}")
  private String staticMemberHost;
  @Value("${tomcat.cluster.static-member-port:}")
  private Integer staticMemberPort;
  @Value("${tomcat.cluster.static-member-unique-id:}")
  private String staticMemberUniqueId;
  @Value("${tomcat.cluster.receiver-address:}")
  private String receiverAddress;
  @Value("${tomcat.cluster.receiver-port:}")
  private Integer receiverPort;
  @Value("${tomcat.cluster.receiver-max-threads:}")
  private Integer receiverMaxThreads;

  @Override
  public void customize( final Context context ) {
    context.setDistributable(true);
    // BackupManager manager = new BackupManager();
    DeltaManager manager = new DeltaManager();
    manager.setExpireSessionsOnShutdown(false);
    manager.setNotifyListenersOnReplication(true);
    context.setManager(manager);
    configureCluster( (Engine)context.getParent().getParent() );
  }

  private void configureCluster(Engine engine) {
    //cluster setting
    SimpleTcpCluster cluster = new SimpleTcpCluster();
    cluster.setChannelStartOptions(3);
    cluster.setChannelSendOptions(8);

    //channel setting
    GroupChannel channel = new GroupChannel();
    StaticMembershipInterceptor interceptor = new StaticMembershipInterceptor();

    // 대상 정보
    StaticMember staticMember = new StaticMember();
    staticMember.setHost(staticMemberHost);
    staticMember.setPort(staticMemberPort);
    staticMember.setSecurePort(-1);
    staticMember.setUniqueId(staticMemberUniqueId);
    interceptor.addStaticMember(staticMember);

    //receiver(현재 자신의 정보)
    NioReceiver receiver = new NioReceiver();
    receiver.setAddress(receiverAddress);
    receiver.setPort(receiverPort);
    receiver.setMaxThreads(receiverMaxThreads);
    channel.setChannelReceiver(receiver);

    channel.addInterceptor(interceptor);

    //sender setting
    ReplicationTransmitter sender = new ReplicationTransmitter();
    sender.setTransport(new PooledParallelSender());
    channel.setChannelSender(sender);

    //interceptor setting
    channel.addInterceptor(new TcpPingInterceptor());
    channel.addInterceptor(new TcpFailureDetector());
    channel.addInterceptor(new MessageDispatchInterceptor());

    cluster.addValve(new ReplicationValve());
    cluster.addValve(new JvmRouteBinderValve());
    cluster.setChannel(channel);
    cluster.addClusterListener(new ClusterSessionListener());
    engine.setCluster(cluster);
  }
}