package com.atlassian.bitbucket.mesh.management;

import com.atlassian.bitbucket.mesh.rpc.v1.RpcSidebandLeaderFragment;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;

import static java.util.Objects.requireNonNull;

class SidebandRequestObserver implements StreamObserver<RpcSidebandLeaderFragment> {

    private static final Logger log = LoggerFactory.getLogger(SidebandRequestObserver.class);

    private final SidebandReceiver receiver;

    SidebandRequestObserver(@Nonnull SidebandReceiver receiver) {
        this.receiver = requireNonNull(receiver, "receiver");
    }

    @Override
    public void onCompleted() {
        receiver.close();
    }

    @Override
    public void onError(Throwable t) {
        Status status = Status.fromThrowable(t);

        //If the RPC was canceled, we treat it as a clean close.
        if (status.getCode() != Status.Code.CANCELLED) {
            //Otherwise, we use whether there's a causal exception to decide whether the error happened
            //locally or remotely
            Throwable cause = status.getCause();
            if (cause == null) {
                //A null cause means the error was sent by the control plane
                log.warn("Sideband channel died", t);
            } else {
                //If a cause is present, it means an exception was thrown locally
                log.warn("Sideband channel failed", cause);
            }
        } else {
            log.debug("Sideband channel canceled");
        }

        receiver.close();
    }

    @Override
    public void onNext(RpcSidebandLeaderFragment fragment) {
        receiver.onReceive(fragment);
    }
}
