package com.atlassian.bitbucket.mesh.repair;

import com.atlassian.bitbucket.mesh.config.GrpcConfig;
import com.atlassian.bitbucket.mesh.coordination.LogicalClock;
import com.atlassian.bitbucket.mesh.git.GitProcessCompletionHandler;
import com.atlassian.bitbucket.mesh.git.Reflog;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.builder.catfile.InteractiveInfoBatchCheckCatFilesStdioHandler;
import com.atlassian.bitbucket.mesh.git.exception.*;
import com.atlassian.bitbucket.mesh.git.gc.GarbageCollectionManager;
import com.atlassian.bitbucket.mesh.git.gc.GcPause;
import com.atlassian.bitbucket.mesh.git.porcelain.GitPorcelain;
import com.atlassian.bitbucket.mesh.git.porcelain.StreamReflogsParameters;
import com.atlassian.bitbucket.mesh.git.porcelain.StreamRefsParameters;
import com.atlassian.bitbucket.mesh.git.ref.Ref;
import com.atlassian.bitbucket.mesh.git.transcode.TranscodeService;
import com.atlassian.bitbucket.mesh.grpc.AbstractFutureResponseObserver;
import com.atlassian.bitbucket.mesh.grpc.BackoffClientCallStreamObserver;
import com.atlassian.bitbucket.mesh.grpc.StdinObserver;
import com.atlassian.bitbucket.mesh.process.OsProcess;
import com.atlassian.bitbucket.mesh.process.ProcessCompletionHandler;
import com.atlassian.bitbucket.mesh.process.ProcessSummary;
import com.atlassian.bitbucket.mesh.process.handler.AbstractProcessStdioHandler;
import com.atlassian.bitbucket.mesh.process.handler.LineDecoder;
import com.atlassian.bitbucket.mesh.repository.*;
import com.atlassian.bitbucket.mesh.rpc.ZeroCopyUtils;
import com.atlassian.bitbucket.mesh.rpc.util.ByteStringUtils;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import com.atlassian.bitbucket.mesh.util.SatiableConsumer;
import com.google.common.io.ByteStreams;
import com.google.protobuf.ByteString;
import com.google.protobuf.UnsafeByteOperations;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.time.Duration;
import java.time.Instant;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.function.Function;

import static com.atlassian.bitbucket.mesh.git.GitConstants.PATH_LOGS;
import static com.atlassian.bitbucket.mesh.gprc.GrpcSupport.getDeadline;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;

/**
 * Implements the {@code RepositoryService/Repair} and {@code RepositoryService/RepairMe} protocols from the side that
 * repairs its local repository replica. For {@code Repair} that is the RPC server side and for {@code RepairMe} it's
 * the RPC client side.
 * <p>
 * The {@code Repair} and {@code RepairMe} flows are nearly identical and differ only in how the flow is initiated and
 * the capabilities are negotiated between the client and server:
 * <ul>
 *     <li>{@code RepairMe}: the client drives the repair and initiates the flow by requesting the repository state
 *     on the server. The client advertises its capabilities as part of the {@link RpcGetRepositoryStateRequest}.
 *     The server calculates the common capabilities and provides them on the {@link RpcGetRepositoryDetailsResponse}.
 *     From this point on, the client requests the information from the server and repairs its local repository.</li>
 *     <li>{@code Repair}: the client <em>initiates</em> the repair process by sending a {@link RpcRepairRequest}.
 *     The server then takes over and drives the repair using the same flow as {@code RepairMe}.</li>
 * </ul>
 *
 * @see RepairTarget
 */
public class RepairSource
        extends AbstractFutureResponseObserver<RpcRepairSourceFragment, RpcRepairTargetFragment, RepairSummary> {

    private static final RepositoryVisitor<Long> GET_VERSION = new RepositoryVisitor<Long>() {

        @Override
        public Long visit(@Nonnull ManagedRepository repository) {
            return repository.getVersion();
        }

        @Override
        public Long defaultValue(Repository repository) {
            return 0L;
        }
    };
    private static final Logger log = LoggerFactory.getLogger(RepairSource.class);

    private final GitProcessBuilderFactory builderFactory;
    private final Instant deadline;
    private final ExecutorService executorService;
    private final GarbageCollectionManager garbageCollectionManager;
    private final GrpcConfig grpcConfig;
    private final LogicalClock logicalClock;
    private final GitPorcelain porcelain;
    private final ReplicaStateRegistry replicaStateRegistry;
    private final RepositoryHashService repositoryHashService;
    private final RepositoryManager repositoryManager;
    //Currently unused, but capability negotiation is implemented in preparation of future protocol changes
    @SuppressWarnings("MismatchedQueryAndUpdateOfCollection")
    private final Set<RepairCapability> sharedCapabilities;
    private final TranscodeService transcodeService;

    private volatile Set<String> advertisedHashes;
    private volatile GcPause gcPause;
    private volatile PackObjects packObjects;
    private volatile StreamObserver<RpcRepairSourceFragment> repairTarget;

    /**
     * Constructor used by {@code RepairMe} by the RPC service implementation.
     */
    public RepairSource(GitProcessBuilderFactory builderFactory, ExecutorService executorService,
                        GarbageCollectionManager garbageCollectionManager, GrpcConfig grpcConfig,
                        LogicalClock logicalClock, GitPorcelain porcelain,
                        StreamObserver<RpcRepairSourceFragment> repairTarget, ReplicaStateRegistry replicaStateRegistry,
                        RepositoryHashService repositoryHashService, RepositoryManager repositoryManager,
                        TranscodeService transcodeService, Duration timeout) {
        this.builderFactory = builderFactory;
        this.executorService = executorService;
        this.garbageCollectionManager = garbageCollectionManager;
        this.grpcConfig = grpcConfig;
        this.logicalClock = logicalClock;
        this.porcelain = porcelain;
        this.repairTarget = repairTarget;
        this.replicaStateRegistry = replicaStateRegistry;
        this.repositoryHashService = repositoryHashService;
        this.repositoryManager = repositoryManager;
        this.transcodeService = transcodeService;

        deadline = getDeadline(timeout);
        sharedCapabilities = EnumSet.noneOf(RepairCapability.class);
    }

    /**
     * Constructor used by {@code Repair} by the RPC client implementation.
     */
    public RepairSource(GitProcessBuilderFactory builderFactory, ExecutorService executorService,
                        GarbageCollectionManager garbageCollectionManager, GrpcConfig grpcConfig,
                        LogicalClock logicalClock, GitPorcelain porcelain, ReplicaStateRegistry replicaStateRegistry,
                        RepositoryHashService repositoryHashService, RepositoryManager repositoryManager,
                        TranscodeService transcodeService, Duration timeout) {
        this(builderFactory, executorService, garbageCollectionManager, grpcConfig, logicalClock, porcelain, null,
                replicaStateRegistry, repositoryHashService, repositoryManager, transcodeService, timeout);
    }

    @Override
    public void beforeStart(ClientCallStreamObserver<RpcRepairSourceFragment> requestStream) {
        ClientCallStreamObserver<RpcRepairSourceFragment> backoffStream =
                new BackoffClientCallStreamObserver<>(requestStream);
        repairTarget = backoffStream;
        super.beforeStart(backoffStream);
    }

    @Override
    public void onCompleted() {
        try {
            repairTarget.onCompleted();
            super.onCompleted();
        } finally {
            wrapUp();
        }
    }

    @Override
    public void onError(Throwable t) {
        try {
            super.onError(t);
        } finally {
            wrapUp();
        }
    }

    @Override
    public void onNext(RpcRepairTargetFragment fragment) {
        logicalClock.onReceived(fragment.getLogicalTimestamp());
        switch (fragment.getCommandOneofCase()) {
            case LIST_REFLOGS:
                onListReflogsRequest(fragment.getListReflogs());
                break;
            case LIST_REFS:
                onListRefsRequest(fragment.getListRefs());
                break;
            case GET_REPOSITORY_DETAILS:
                onRepositoryDetailsRequest(fragment.getGetRepositoryDetails());
                break;
            case GET_REPOSITORY_STATE:
                onRepositoryStateRequest(fragment.getGetRepositoryState());
                break;
            case GET_PACK:
                onPackObjectsFragment(fragment.getGetPack());
                break;
            case GET_REFLOGS:
                onGetReflogs(fragment.getGetReflogs());
                break;
            case SUMMARY:
                onSummary(fragment.getSummary());
                break;
            default:
                log.warn("Unsupported fragment of type {}", fragment.getCommandOneofCase());
        }
    }

    //Only called by MeshClient.repair to kickstart the repair by letting the server know which repository to repair.
    public void startRepair(@Nonnull String sourceRepositoryId, @Nonnull String targetRepositoryId) {
        requireNonNull(sourceRepositoryId, "sourceRepositoryId");
        requireNonNull(targetRepositoryId, "targetRepositoryId");

        log.debug("[{}] Instructing replica to start repair", targetRepositoryId);

        //Start the flow by instructing the repair target which repository to repair
        sendFragmentToTarget(fragment -> fragment.setRepairRequest(RpcRepairRequest.newBuilder()
                .setSourceRepository(sourceRepositoryId)
                .setRepository(targetRepositoryId)));
    }

    private static Path getReflogPath(Repository repository, String refId) {
        Path result = repository.getPath().resolve(PATH_LOGS).resolve(refId);
        if (Files.isRegularFile(result)) {
            return result;
        }

        if (repository instanceof ManagedRepository) {
            ManagedRepository managed = (ManagedRepository) repository;
            result = managed.getSharedObjects().getReflogDir(managed).resolve(refId);
            return Files.isRegularFile(result) ? result : null;
        }
        return null;
    }

    private Duration getTimeout() {
        return Duration.between(Instant.now(), deadline);
    }

    private void onGetReflogs(RpcGetReflogsRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());
        log.debug("[{}] Returning {} reflogs", repository, request.getPathsCount());

        //Create a deep copy to avoid data corruption if the Netty buffer backing request is reused for
        //other messages while the async task is sending reflogs.
        RpcGetReflogsRequest safeRequest = ZeroCopyUtils.deepCopy(request);
        executorService.submit(() -> {
            byte[] buffer = new byte[64 * 1024];
            for (ByteString pathBytes : safeRequest.getPathsList()) {
                sendReflogFile(repository, pathBytes, buffer);
            }
        });
    }

    private void onListReflogsRequest(RpcListReflogsRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());
        if (gcPause == null) {
            //Because repair accesses reflog files in the repository directly, we need to hold the GC lock to ensure
            //they're stable.
            gcPause = garbageCollectionManager.pause(repository, getTimeout());
            if (gcPause == null) {
                log.warn("[{}] Cannot stream reflogs; the GC lock could not be acquired", repository);
                throw new RepositoryRepairFailedException("Failed to acquire GC lock");
            }
        }

        log.debug("[{}] Streaming reflogs", repository);
        executorService.submit(() -> {
            StreamReflogsCallback callback = new StreamReflogsCallback();
            try {
                porcelain.streamReflogs(new StreamReflogsParameters.Builder(repository)
                        .timeout(getTimeout().getSeconds())
                        .build(), callback);
                callback.sendLastChunk();
                log.debug("[{}] Streamed {} reflogs", repository, callback.total);
            } catch (IOException e) {
                log.warn("[{}] Failed to stream reflogs", repository, e);
                repairTarget.onError(new RepositoryRepairFailedException("Failed to stream reflogs", e));
            }
        });
    }

    private void onListRefsRequest(RpcListRefsRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());
        log.debug("[{}] Streaming refs", repository);

        executorService.submit(() -> {
            StreamRefsCallback callback = new StreamRefsCallback();
            try {
                porcelain.streamRefs(new StreamRefsParameters.Builder(repository)
                        .timeout(getTimeout().getSeconds())
                        .darkRefs(true)
                        .publicRefs(true)
                        .stashRefs(true)
                        .build(), callback);
                callback.sendLastChunk();
                log.debug("[{}] Streamed {} refs", repository, callback.total);
            } catch (IOException e) {
                log.warn("[{}] Failed to stream refs", repository, e);
                repairTarget.onError(new RepositoryRepairFailedException("Failed to stream refs", e));
            }
        });
    }

    private void onPackObjectsFragment(RpcPackObjectsStdinFragment packObjectsFragment) {
        switch (packObjectsFragment.getRequestOneofCase()) {
            case REQUEST:
                if (packObjects == null) {
                    RpcPackObjectsRequest request = packObjectsFragment.getRequest();
                    if (request.getRevsCount() > 0) {
                        repairTarget.onError(new FailedPreconditionException(
                                "Use stdin fragments to provide pack-objects revs, not the revs field!"));
                    }
                    packObjects = new PackObjects(request);
                } else {
                    repairTarget.onError(new FailedPreconditionException("Pack objects has already been started"));
                }
                break;
            case STDIN:
                if (packObjects == null) {
                    repairTarget.onError(new FailedPreconditionException("Received stdin before pack-objects request"));
                } else {
                    packObjects.onNext(packObjectsFragment.getStdin());
                }
                break;
        }
    }

    private void onReflogNotFound(Repository repository, ByteString pathBytes, String relPath) {
        log.debug("[{}] Requested reflog {} not found", repository, relPath);
        sendFragmentToTarget(fragment -> fragment.setReflogStart(RpcReflogStart.newBuilder()
                .setPath(pathBytes)
                .setNotFound(true)));
    }

    private void onRepositoryDetailsRequest(RpcGetRepositoryDetailsRequest request) {
        Repository repository = repositoryManager.getById(request.getRepository());
        log.debug("[{}] Returning repository details", repository);

        RpcGetRepositoryDetailsResponse.Builder builder = RpcGetRepositoryDetailsResponse.newBuilder()
                .putAllMetadata(repositoryManager.getMetadata(repository))
                .setTranscodeEnabled(transcodeService.isEnabled(repository));
        String defaultBranch = repositoryManager.getDefaultBranch(repository);
        if (defaultBranch != null) {
            builder.setDefaultBranch(ByteString.copyFromUtf8(defaultBranch));
        }
        sendFragmentToTarget(fragment -> fragment.setRepositoryDetails(builder));
    }

    private void onRepositoryStateRequest(RpcGetRepositoryStateRequest request) {
        Repository repository = repositoryManager.findById(request.getRepository());

        //Process the advertised capabilities and reply with the shared capabilities
        RpcRepairCapabilitiesAdvertisement.Builder capabilities = RpcRepairCapabilitiesAdvertisement.newBuilder();
        for (String advertised : request.getCapabilities().getCapabilitiesList()) {
            RepairCapability.fromId(advertised).ifPresent(capability -> {
                sharedCapabilities.add(capability);
                capabilities.addCapabilities(advertised);
            });
        }

        RpcGetRepositoryStateResponse.Builder builder = RpcGetRepositoryStateResponse.newBuilder()
                .setCapabilities(capabilities);
        if (repository == null) {
            builder.setState(RpcReplicaState.REPLICA_STATE_MISSING);
        } else {
            //TODO: add last known transactionId if/when we start tracking it
            if (request.getNoCache()) {
                log.debug("[{}] Clearing cached repository hashes", repository);
                repositoryHashService.resetContentHash(repository);
                repositoryHashService.resetMetadataHash(repository);
            }
            String contentHash = repositoryHashService.getContentHash(repository);
            String metadataHash = repositoryHashService.getMetadataHash(repository);
            ReplicaState state = replicaStateRegistry.getState(repository);
            log.debug("[{}] Sending repository state: {} (Metadata: {}, Content: {})",
                    repository, state, metadataHash, contentHash);

            builder.setState(state.toRpc())
                    .setContentHash(contentHash)
                    .setMetadataHash(metadataHash)
                    .setVersion(repository.accept(GET_VERSION));
        }

        sendFragmentToTarget(fragment -> fragment.setRepositoryState(builder));
    }

    private void onSummary(RpcRepairSummary summary) {
        setResult(RepairSummary.fromRepairSummary(summary));
    }

    private void sendFragmentToTarget(Function<RpcRepairSourceFragment.Builder, RpcRepairSourceFragment.Builder> builder) {
        repairTarget.onNext(builder.apply(RpcRepairSourceFragment.newBuilder()
                .setLogicalTimestamp(logicalClock.tick()))
                .build());
    }

    private void sendReflogFile(Repository repository, ByteString pathBytes, byte[] buffer) {
        String relPath = pathBytes.toStringUtf8();
        Path reflogPath = getReflogPath(repository, relPath);
        if (reflogPath == null) {
            onReflogNotFound(repository, pathBytes, relPath);
            return;
        }

        try (InputStream input = Files.newInputStream(reflogPath)) {
            sendFragmentToTarget(fragment -> fragment.setReflogStart(RpcReflogStart.newBuilder()
                    .setPath(pathBytes)));

            boolean eof = false;
            int read;
            //Fully read the buffer to reduce the number of chunks sent. Fewer larger messages is slightly faster.
            while ((read = ByteStreams.read(input, buffer, 0, buffer.length)) > 0) {
                eof = read < buffer.length;
                RpcStreamChunk.Builder chunk = RpcStreamChunk.newBuilder()
                        .setData(UnsafeByteOperations.unsafeWrap(buffer, 0, read))
                        .setClosed(eof);
                sendFragmentToTarget(fragment -> fragment.setReflogChunk(chunk));
            }
            if (!eof) {
                //No chunk with the closed flag has been sent because the last chunk was exactly the buffer length
                sendFragmentToTarget(fragment -> fragment.setReflogChunk(RpcStreamChunk.newBuilder()
                                .setData(ByteString.EMPTY)
                                .setClosed(true)));
            }
        } catch (FileNotFoundException | NoSuchFileException e) {
            onReflogNotFound(repository, pathBytes, relPath);
        } catch (IOException e) {
            throw new RepositoryRepairFailedException(
                    "Failed to read reflog " + relPath + " for repository " + repository, e);
        }
    }

    private void wrapUp() {
        try (GcPause ignored = gcPause) {
            if (packObjects != null) {
                packObjects.close();
            }
        }
    }

    private class PackObjects {

        private final Future<Void> future;
        private final PackObjectsStdioHandler handler;
        private final LineDecoder lineDecoder;
        private final Repository repository;

        private volatile Future<Void> catFilesFuture;
        private volatile InteractiveInfoBatchCheckCatFilesStdioHandler catFilesHandler;

        PackObjects(RpcPackObjectsRequest request) {
            log.debug("[{}] Processing pack request", request.getRepository());
            repository = repositoryManager.getById(request.getRepository());
            lineDecoder = new LineDecoder(UTF_8);
            handler = new PackObjectsStdioHandler(repository);
            future = builderFactory.builder(request).packObjects()
                    .withTimeout(getTimeout())
                    .all(request.getAll())
                    .deltaBaseOffset(true)
                    .revs(true)
                    .thin(request.getThin())
                    .build(handler)
                    .start();
        }

        void close() {
            future.cancel(true);
            if (catFilesHandler != null) {
                catFilesHandler.close();
            }
            if (catFilesFuture != null) {
                catFilesFuture.cancel(true);
            }
        }

        void onNext(RpcStreamChunk stdin) {
            for (ByteBuffer buffer : stdin.getData().asReadOnlyByteBufferList()) {
                lineDecoder.forEach(buffer, false, this::onStdinLine);
            }

            if (stdin.getClosed()) {
                lineDecoder.forEach(ByteString.EMPTY.asReadOnlyByteBuffer(), true, this::onStdinLine);
                if (catFilesHandler != null) {
                    catFilesHandler.close();
                }
                handler.close();
            }
        }

        private boolean onStdinLine(String line, boolean truncated) {
            try {
                //pack-objects also supports --not and --shallow <sha> arguments
                if (line.startsWith("--")) {
                    handler.writeLine(line, false);
                } else {
                    String rev = (line.length() > 1 && line.charAt(0) == '^') ? line.substring(1) : line;
                    if (advertisedHashes != null && advertisedHashes.contains(rev)) {
                        //We've advertised the revision, so no need to check whether it exists locally
                        handler.writeLine(line, false);
                    } else {
                        if (catFilesFuture == null) {
                            catFilesHandler = new InteractiveInfoBatchCheckCatFilesStdioHandler();
                            catFilesFuture = builderFactory.builder(repository).catFile()
                                    .batchCheck()
                                    .withTimeout(getTimeout())
                                    .build(catFilesHandler)
                                    .start();
                        }
                        catFilesHandler.getInfo(rev).ifPresent(info -> handler.writeLine(line, false));
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.warn("[{}] Interrupted while resolving commit IDs", repository);
                throw new AbortedException("Interrupted while resolving commit IDs", e);
            }

            return true;
        }
    }

    private class PackObjectsStdioHandler extends AbstractProcessStdioHandler<Void> implements ProcessCompletionHandler {

        private final Repository repository;
        private final StdinObserver stdinObserver;

        PackObjectsStdioHandler(Repository repository) {
            this.repository = repository;

            stdinObserver = new StdinObserver();
        }

        @Override
        public Void getResult() {
            return null;
        }

        @Override
        public void onComplete(@Nonnull ProcessSummary summary) {
            try {
                new GitProcessCompletionHandler().onComplete(summary);
            } catch (RuntimeException e) {
                log.warn("[{}] git-pack objects failed", repository, e);
                if (e instanceof AbstractGitException) {
                    repairTarget.onError(((AbstractGitException) e).toStatusException());
                } else if (e instanceof StatusRuntimeException) {
                    repairTarget.onError(e);
                } else {
                    repairTarget.onError(new CommandFailedException("git pack-objects failed", e, summary));
                }
            }
        }

        @Override
        public void onStart(@Nonnull OsProcess process) {
            super.onStart(process);
            stdinObserver.onStart(process);
        }

        @Override
        public boolean onStdinReady(@Nonnull ByteBuffer buffer) {
            return stdinObserver.onStdinReady(buffer);
        }

        @Override
        public void onStdout(@Nonnull ByteBuffer buffer, boolean closed) {
            sendFragmentToTarget(fragment -> fragment.setPackChunk(RpcStreamChunk.newBuilder()
                    .setClosed(closed)
                    .setData(ByteStringUtils.unsafeWrap(buffer))));
        }

        void close() {
            stdinObserver.closeStdin();
        }

        void writeLine(String line, boolean closed) {
            stdinObserver.writeStdin(ByteString.copyFromUtf8(line + '\n'), closed);
        }
    }

    private class StreamRefsCallback implements SatiableConsumer<Ref> {

        private final RpcRefsChunk.Builder chunk = RpcRefsChunk.newBuilder();

        private volatile long chunkSize;
        private volatile long total;

        @Override
        public boolean accept(Ref ref) {
            queueRef(ref);
            if (advertisedHashes == null) {
                advertisedHashes = new HashSet<>(256);
            }
            if (advertisedHashes.size() < 1024 && !ref.isDark()) {
                //Remember some of the refs we've advertised so we don't have to check whether the objects exist
                //locally when providing a pack of missing objects later on in the repair process. Dark refs are
                //not validated by git, so for those we can't trust that the referenced object exists.
                advertisedHashes.add(ref.getLatestCommit());
            }
            ++total;

            return false;
        }

        public void sendLastChunk() {
            emitChunk(false);
        }

        private void emitChunk(boolean more) {
            chunk.setMoreResponses(more);
            sendFragmentToTarget(fragment -> fragment.setListRefsChunk(chunk));

            chunk.clear();
            chunkSize = 0L;
        }

        private void queueRef(Ref ref) {
            RpcRef rpcRef = RpcRef.newBuilder()
                    .setObjectId(ref.getObjectId())
                    .setId(ByteString.copyFromUtf8(ref.getId()))
                    .build();
            long refSize = rpcRef.getSerializedSize();
            if (chunkSize + refSize > grpcConfig.getMaxFragmentSize()) {
                emitChunk(true);
            }
            chunk.addRefs(rpcRef);
            chunkSize += refSize;
        }
    }

    private class StreamReflogsCallback implements SatiableConsumer<Reflog> {

        private final RpcReflogsChunk.Builder chunk = RpcReflogsChunk.newBuilder();

        private volatile long chunkSize;
        private volatile long total;

        @Override
        public boolean accept(Reflog reflog) {
            RpcOsFile osFile = reflog.toOsFile();
            long size = osFile.getSerializedSize();
            if (chunkSize + size > grpcConfig.getMaxFragmentSize()) {
                emitChunk(true);
            }
            chunk.addFiles(reflog.toOsFile());
            chunkSize += size;
            ++total;

            return false;
        }

        public void sendLastChunk() {
            emitChunk(false);
        }

        private void emitChunk(boolean more) {
            chunk.setMoreResponses(more);
            sendFragmentToTarget(fragment -> fragment.setListReflogsChunk(chunk));

            chunk.clear();
            chunkSize = 0L;
        }
    }
}
