package com.atlassian.bitbucket.mesh.gprc;

import io.grpc.stub.StreamObserver;

/**
 * {@link StreamObserver} wrapper that adds synchronization to the methods that require it. This wrapper only needs to
 * be used when interactions with the wrapped {@link StreamObserver} can be triggered concurrently from different
 * threads. <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 StreamObserver} from
 * callbacks on the incoming {@link StreamObserver}, you don't need this wrapping.
 *
 * @see StreamObserver
 * @param <V> the message type
 */
public class SynchronizedStreamObserver<V> implements StreamObserver<V> {

    private final StreamObserver<V> delegate;
    private final Object lock;

    public SynchronizedStreamObserver(StreamObserver<V> delegate) {
        this.delegate = delegate;
        lock = new Object();
    }

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

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

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