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.Channel;
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;

  @Override
  public void customize( final Context context ) {
    context.setDistributable(true);

    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(Channel.SND_RX_SEQ|Channel.SND_TX_SEQ);
    cluster.setChannelSendOptions(Channel.SEND_OPTIONS_ASYNCHRONOUS);

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

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

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

    channel.addInterceptor(interceptor);

/*    cluster.setChannelSendOptions(6);

    //channel
    GroupChannel channel = new GroupChannel();
    //membership setting
    McastService mcastService = new McastService();
    mcastService.setAddress("228.0.0.4");
    mcastService.setPort(45564); // TCP&UDP port 오픈 필요
    mcastService.setFrequency(500);
    mcastService.setDropTime(3000);
    channel.setMembershipService(mcastService);

    //receiver
    NioReceiver receiver = new NioReceiver();
    receiver.setAddress("auto");
    receiver.setMaxThreads(6);
    receiver.setPort(5000); // TCP port 오픈 필요
    channel.setChannelReceiver(receiver);
*/

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

    //interceptor
    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);
  }
}