package com.atlassian.bitbucket.mesh.gprc;

import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.StreamObserver;

/**
 * {@link ClientCallStreamObserver} wrapper that adds synchronization to the methods that require it. This wrapper
 * only needs to be used when interactions with the wrapped {@link ClientCallStreamObserver} can be triggered
 * concurrently from different threads.
 * <p>
 * <em>Usually</em> this synchronization is not necessary because gRPC applies synchronization to the
 * incoming {@link StreamObserver}. In other words, if you only interact with the outgoing
 * {@link ClientCallStreamObserver} from callbacks on the incoming {@link StreamObserver}, you don't need this wrapping.
 *
 * @see ClientCallStreamObserver
 * @see StreamObserver
 * @param <ReqT> the message type
 */
public class SynchronizedClientCallStreamObserver<ReqT> extends ClientCallStreamObserver<ReqT> {

    private final ClientCallStreamObserver<ReqT> delegate;
    private final Object lock;

    public SynchronizedClientCallStreamObserver(ClientCallStreamObserver<ReqT> delegate) {
        this.delegate = delegate;
        lock = new Object();
    }

    @Override
    public void cancel(String message, Throwable cause) {
        synchronized (lock) {
            delegate.cancel(message, cause);
        }
    }

    @Override
    public void disableAutoInboundFlowControl() {
        //Not synchronized because it is only allowed to be called from beforeStart
        delegate.disableAutoInboundFlowControl();
    }

    @Override
    public void disableAutoRequestWithInitial(int request) {
        //Not synchronized because it is only allowed to be called from beforeStart
        delegate.disableAutoRequestWithInitial(request);
    }

    @Override
    public boolean isReady() {
        //Not synchronized because, although not documented as such, the implementation is thread-safe
        return delegate.isReady();
    }

    @Override
    public void onCompleted() {
        synchronized (lock) {
            delegate.onCompleted();
        }
    }

    @Override
    public void onError(Throwable t) {
        synchronized (lock) {
            delegate.onError(t);
        }
    }

    @Override
    public void onNext(ReqT value) {
        synchronized (lock) {
            delegate.onNext(value);
        }
    }

    @Override
    public void request(int count) {
        //Not synchronized because it documented as safe to call from different threads
        delegate.request(count);
    }

    @Override
    public void setMessageCompression(boolean enable) {
        synchronized (lock) {
            delegate.setMessageCompression(enable);
        }
    }

    @Override
    public void setOnReadyHandler(Runnable onReadyHandler) {
        //Not synchronized because it is only allowed to be called from beforeStart
        delegate.setOnReadyHandler(onReadyHandler);
    }
}
