package com.atlassian.bitbucket.mesh.grpc;

import com.atlassian.bitbucket.mesh.process.ProcessCompletionHandler;
import com.atlassian.bitbucket.mesh.process.ProcessStdioHandler;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcStreamFragment;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.ClientResponseObserver;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.nio.ByteBuffer;

/**
 * A {@link ClientResponseObserver} implementation which receives {@link RpcStreamFragment}s and forwards them to
 * a provided {@link ProcessStdioHandler}.
 * <p>
 * Note: This observer is only intended for the client side of server-side streaming operations. It doesn't
 * support providing {@code stdin}, so it's not appropriate for bidirectional or client-side streaming, nor
 * can it be used for unary requests (which will provide a single pre-parsed result, rather than a series of
 * {@link RpcStreamFragment}s).
 *
 * @param <T> the type of result returned by the handler, if the request completes successfully
 */
public class FragmentResponseObserver<T> extends AbstractFragmentResponseObserver<Object, T> {

    /**
     * Creates a new {@code FragmentResponseObserver} which uses a {@link ProcessStdioHandler} that also implements
     * {@link ProcessCompletionHandler}.
     *
     * @param stdioHandler the I/O <i>and completion</i> handler
     * @param <H> the handler's type
     */
    public <H extends ProcessStdioHandler<T> & ProcessCompletionHandler> FragmentResponseObserver(
            @Nonnull H stdioHandler) {
        super(stdioHandler);
    }

    /**
     * Creates a new {@code FragmentResponseObserver} which uses the provided {@link ProcessStdioHandler} and
     * {@link ProcessCompletionHandler}. If the provided I/O handler implements {@link ProcessCompletionHandler}
     * the separate handler may be {@code null}; otherwise, an explicit completion handler must be provided.
     *
     * @param stdioHandler      the I/O handler, which may also be a completion handler
     * @param completionHandler the completion handler, if the I/O handler is not one
     */
    public FragmentResponseObserver(@Nonnull ProcessStdioHandler<T> stdioHandler,
                                    @Nullable ProcessCompletionHandler completionHandler) {
        super(stdioHandler, completionHandler);
    }

    @Override
    protected RemoteProcess<Object> createProcess(ClientCallStreamObserver<Object> requestStream) {
        return new ReadOnlyRemoteProcess(requestStream);
    }

    /**
     * A read-only {@link RemoteProcess} implementation which does not allow writing to the remote process.
     */
    private static class ReadOnlyRemoteProcess extends RemoteProcess<Object> {

        ReadOnlyRemoteProcess(ClientCallStreamObserver<Object> requestStream) {
            super(requestStream);
        }

        @Override
        public void closeStdin(boolean force) {
            //No-op; we can't dispatch stdin. This doesn't throw an exception to allow for output-only handlers
            //that apply a "safety" closeStdin to ensure the process doesn't get stuck waiting
        }

        @Override
        public boolean hasPendingWrites() {
            //For the purposes of handlers used with this observer, there are never any pending writes
            return false;
        }

        @Override
        public void wantWrite() {
            throw new UnsupportedOperationException("Bidirectional streaming is not supported by this observer");
        }

        @Override
        public void writeStdin(@Nonnull ByteBuffer buffer) {
            throw new UnsupportedOperationException("Bidirectional streaming is not supported by this observer");
        }
    }
}
