package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.git.commit.Commit;
import com.atlassian.bitbucket.mesh.git.commit.CommitCallback;
import com.atlassian.bitbucket.mesh.git.content.MinimalChange;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcGetCommitsResponseFragment;
import com.google.common.collect.Maps;
import io.grpc.stub.StreamObserver;

import javax.annotation.Nonnull;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

class ObserverCommitCallback
        extends AbstractObserverCallback<RpcGetCommitsResponseFragment>
        implements CommitCallback {

    private final List<Commit> commits;
    private final Map<String, MinimalChange> renames;

    ObserverCommitCallback(StreamObserver<RpcGetCommitsResponseFragment> observer) {
        super(observer);

        commits = new ArrayList<>(FRAGMENT_SIZE);
        renames = new HashMap<>();
    }

    @Override
    public boolean onCommit(@Nonnull Commit commit, MinimalChange renameChange) {
        commits.add(commit);
        if (renameChange != null) {
            renames.put(commit.getId(), renameChange);
        }
        if (commits.size() >= FRAGMENT_SIZE) {
            sendFragment();
        }

        return !isCanceled();
    }

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

    private void sendFragment() {
        RpcGetCommitsResponseFragment.Builder builder = RpcGetCommitsResponseFragment.newBuilder();

        commits.stream().map(Commit::toCommit).forEach(builder::addCommits);
        commits.clear();

        builder.putAllRenameChanges(Maps.transformValues(renames, MinimalChange::toChange));
        renames.clear();

        observer.onNext(builder.build());
    }

    private void sendLastFragment() {
        if (!commits.isEmpty()) {
            sendFragment();
        }
    }
}
