package com.atlassian.stash.internal.cluster;

import com.atlassian.util.contentcache.internal.util.PipedStreams;
import com.google.common.io.Closeables;
import com.hazelcast.core.HazelcastInstance;
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 javax.annotation.Nonnull;
import java.io.*;

import static com.google.common.base.Preconditions.checkState;

/**
* {@link ClusterJoinRequest} implementation for unit testing join checks
*/
public class PipedClusterJoinRequest implements ClusterJoinRequest, Closeable {

    private final HazelcastInstance hazelcast;
    private final ClusterJoinMode joinMode;

    private ObjectDataInputStream input;
    private ObjectDataOutputStream output;

    public PipedClusterJoinRequest(HazelcastInstance hazelcast, ClusterJoinMode joinMode) {
        this.joinMode = joinMode;
        this.hazelcast = hazelcast;
    }

    public PipedClusterJoinRequest(HazelcastInstance hazelcast, SerializationService serializationService, PipedClusterJoinRequest other) throws IOException {
        checkState(other.input == null, "Pipe already connected");
        this.hazelcast = hazelcast;
        this.joinMode = other.joinMode == ClusterJoinMode.ACCEPT ? ClusterJoinMode.CONNECT : ClusterJoinMode.ACCEPT;

        // Connect streams one way
        connect(serializationService, this, other);
        // And now back the other way
        connect(serializationService, other, this);
    }

    private void connect(SerializationService serializationService, PipedClusterJoinRequest source, PipedClusterJoinRequest sink) throws IOException {
        PipedStreams streams = new PipedStreams(32 * 1024);

        source.input = serializationService.createObjectDataInputStream(streams.input());
        sink.output = serializationService.createObjectDataOutputStream(streams.output());
    }

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

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

    @Nonnull
    @Override
    public ObjectDataInput in() {
        checkState(input != null, "Pipe not connected");
        return input;
    }

    @Nonnull
    @Override
    public ObjectDataOutput out() {
        checkState(output != null, "Pipe not connected");
        return output;
    }

    @Override
    public void close() throws IOException {
        Closeables.close(input, true);
        Closeables.close(output, true);
    }
}
