package com.atlassian.stash.internal.hazelcast;

import com.atlassian.stash.internal.cluster.ClusterJoinManager;
import com.atlassian.stash.internal.cluster.ClusterJoinMode;
import com.atlassian.stash.internal.cluster.ClusterJoinRequest;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.HazelcastInstanceAware;
import com.hazelcast.instance.HazelcastInstanceImpl;
import com.hazelcast.instance.Node;
import com.hazelcast.nio.MemberSocketInterceptor;
import com.hazelcast.nio.ObjectDataInput;
import com.hazelcast.nio.ObjectDataOutput;
import com.hazelcast.nio.serialization.ObjectDataInputStream;
import com.hazelcast.nio.serialization.ObjectDataOutputStream;
import com.hazelcast.nio.serialization.SerializationService;
import com.hazelcast.spring.context.SpringAware;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Nonnull;
import java.io.IOException;
import java.net.Socket;
import java.util.Properties;

/**
 * A {@link MemberSocketInterceptor} which delegates to {@link ClusterJoinManager} to verify if a member can join.
 * <p>
 * Unfortunately this cannot be a singleton as it requires access to the {@link Node node's} {@link SerializationService}
 *
 * @since 3.3
 */
@SpringAware
public class ClusterJoinSocketInterceptor implements MemberSocketInterceptor, HazelcastInstanceAware {

    private ClusterJoinManager clusterJoinManager;
    private HazelcastInstanceImpl hazelcast;

    @Override
    public void init(Properties properties) {
        // no-op
    }

    @Override
    public void onAccept(Socket socket) throws IOException {
        clusterJoinManager.accept(new SocketClusterJoinRequest(socket, hazelcast, ClusterJoinMode.ACCEPT));
    }

    @Override
    public void onConnect(Socket socket) throws IOException {
        clusterJoinManager.connect(new SocketClusterJoinRequest(socket, hazelcast, ClusterJoinMode.CONNECT));
    }

    @Autowired
    public void setClusterJoinManager(ClusterJoinManager clusterJoinManager) {
        this.clusterJoinManager = clusterJoinManager;
    }

    @Override
    public void setHazelcastInstance(HazelcastInstance hazelcast) {
        this.hazelcast = (HazelcastInstanceImpl) hazelcast;
    }

    static class SocketClusterJoinRequest implements ClusterJoinRequest {

        private final ClusterJoinMode joinMode;
        private final HazelcastInstanceImpl hazelcast;
        private final ObjectDataInputStream in;
        private final ObjectDataOutputStream out;

        SocketClusterJoinRequest(Socket socket, HazelcastInstanceImpl hazelcast, ClusterJoinMode joinMode) throws IOException {
            this.joinMode = joinMode;
            this.hazelcast = hazelcast;

            in = hazelcast.getSerializationService().createObjectDataInputStream(socket.getInputStream());
            out = hazelcast.getSerializationService().createObjectDataOutputStream(socket.getOutputStream());
        }

        @Nonnull
        @Override
        public HazelcastInstance getHazelcast() {
            return hazelcast;
        }

        @Nonnull
        @Override
        public ClusterJoinMode getJoinMode() {
            return joinMode;
        }

        @Nonnull
        @Override
        public ObjectDataInput in() {
            return in;
        }

        @Nonnull
        @Override
        public ObjectDataOutput out() {
            return out;
        }
    }
}
