package com.atlassian.bitbucket.mesh.management;

import com.atlassian.bitbucket.mesh.repository.RepositoryAttributeManager;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcReplaceReadOnlyRepositoriesFragment;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcReplaceReadOnlyRepositoriesResponse;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.util.HashSet;
import java.util.Set;

public class ReplaceReadOnlyRepositoriesRequestObserver implements StreamObserver<RpcReplaceReadOnlyRepositoriesFragment> {

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

    private final Set<String> readOnlyRepositories;
    private final RepositoryAttributeManager repositoryAttributeManager;
    private final StreamObserver<RpcReplaceReadOnlyRepositoriesResponse> responseObserver;

    public ReplaceReadOnlyRepositoriesRequestObserver(@Nonnull StreamObserver<RpcReplaceReadOnlyRepositoriesResponse> responseObserver,
                                                      @Nonnull RepositoryAttributeManager repositoryAttributeManager) {
        this.responseObserver = responseObserver;
        this.repositoryAttributeManager = repositoryAttributeManager;

        readOnlyRepositories = new HashSet<>();
    }

    @Override
    public void onNext(RpcReplaceReadOnlyRepositoriesFragment fragment) {
        readOnlyRepositories.addAll(fragment.getRepositoriesList());
    }

    @Override
    public void onError(Throwable t) {
        if (Status.fromThrowable(t).getCode() == Status.Code.CANCELLED) {
            // Log cancellation more quietly; it's not an error for the client to cancel
            log.debug("The client has canceled the request");
        } else {
            log.error("Client error", t);
        }

        //If the client canceled the request, or had some other failure, it is still necessary to complete
        //the server-side observer to prevent the RPC from lingering until its deadline
        responseObserver.onNext(RpcReplaceReadOnlyRepositoriesResponse.getDefaultInstance());
        responseObserver.onCompleted();
    }

    @Override
    public void onCompleted() {
        repositoryAttributeManager.replaceReadOnly(readOnlyRepositories);
        responseObserver.onNext(RpcReplaceReadOnlyRepositoriesResponse.getDefaultInstance());
        responseObserver.onCompleted();
    }
}