package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.git.commit.CommitGraphNode;
import com.atlassian.bitbucket.mesh.git.commit.TraversalCallback;
import com.atlassian.bitbucket.mesh.git.commit.TraversalStatus;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcMinimalCommit;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcTraverseCommitsResponseFragment;
import io.grpc.stub.StreamObserver;

import javax.annotation.Nonnull;

class ObserverTraversalCallback
        extends AbstractObserverCallback<RpcTraverseCommitsResponseFragment>
        implements TraversalCallback {

    private final RpcTraverseCommitsResponseFragment.Builder fragment;

    ObserverTraversalCallback(StreamObserver<RpcTraverseCommitsResponseFragment> observer) {
        super(observer);

        fragment = RpcTraverseCommitsResponseFragment.newBuilder();
    }

    public void onEnd() {
        sendLastFragment();
        observer.onCompleted();
    }

    @Nonnull
    @Override
    public TraversalStatus onNode(@Nonnull CommitGraphNode node) {
        fragment.addCommits(RpcMinimalCommit.newBuilder()
                .setId(node.getId())
                .addAllParents(node.getParentIds()));
        if (fragment.getCommitsCount() >= FRAGMENT_SIZE) {
            sendFragment();
        }
        return TraversalStatus.CONTINUE;
    }

    private void sendFragment() {
        observer.onNext(fragment.build());
        fragment.clearCommits();
    }

    private void sendLastFragment() {
        if (fragment.getCommitsCount() > 0) {
            sendFragment();
        }
    }
}
