package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.clock.TickingClock;
import com.atlassian.bitbucket.mesh.process.ProcessSummary;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcProgressUpdate;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcPushProgressFragment;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcPushResponseFragment;
import io.grpc.stub.StreamObserver;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;

import java.time.Duration;

import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;

@RunWith(MockitoJUnitRunner.class)
public class ObserverPushCallbackTest {

    private TickingClock clock;
    @Mock
    private StreamObserver<RpcPushResponseFragment> observer;
    private ObserverPushCallback underTest;

    @Before
    public void setup() {
        clock = new TickingClock();
        // 4 Hz - send message every 250ms
        underTest = new ObserverPushCallback(65536, 4f, observer, clock);
    }

    @Test
    public void testThrottledProgressUpdates() {
        // First call is always flushed immediately
        underTest.onProgress(0, "message");
        // Send 3 messages every 100ms - the third message is over the 250ms threshold and updates are flushed
        clock.tick(Duration.ofMillis(100));
        underTest.onProgress(1, "message");
        clock.tick(Duration.ofMillis(100));
        underTest.onProgress(2, "message");
        clock.tick(Duration.ofMillis(100));
        underTest.onProgress(3, "message");
        // Send another 3 messages every 100ms - the 6th message is sent at 600ms total time, which is again 50ms after
        // the last flush and updates are flushed
        clock.tick(Duration.ofMillis(100));
        underTest.onProgress(4, "message");
        clock.tick(Duration.ofMillis(100));
        underTest.onProgress(5, "message");
        clock.tick(Duration.ofMillis(100));
        underTest.onProgress(6, "message");
        // Send another message, but this time flushing should happen because the command ended
        clock.tick(Duration.ofMillis(100));
        underTest.onProgress(7, "message");
        underTest.onEnd(new ProcessSummary.Builder("").build());

        verify(observer).onNext(RpcPushResponseFragment.newBuilder()
                .setProgressUpdate(RpcPushProgressFragment.newBuilder()
                        .addProgressUpdates(RpcProgressUpdate.newBuilder().setPercentage(0).setMessage("message").build())
                        .build())
                .build());
        verify(observer).onNext(RpcPushResponseFragment.newBuilder()
                .setProgressUpdate(RpcPushProgressFragment.newBuilder()
                        .addProgressUpdates(RpcProgressUpdate.newBuilder().setPercentage(1).setMessage("message").build())
                        .addProgressUpdates(RpcProgressUpdate.newBuilder().setPercentage(2).setMessage("message").build())
                        .addProgressUpdates(RpcProgressUpdate.newBuilder().setPercentage(3).setMessage("message").build())
                        .build())
                .build());
        verify(observer).onNext(RpcPushResponseFragment.newBuilder()
                .setProgressUpdate(RpcPushProgressFragment.newBuilder()
                        .addProgressUpdates(RpcProgressUpdate.newBuilder().setPercentage(4).setMessage("message").build())
                        .addProgressUpdates(RpcProgressUpdate.newBuilder().setPercentage(5).setMessage("message").build())
                        .addProgressUpdates(RpcProgressUpdate.newBuilder().setPercentage(6).setMessage("message").build())
                        .build())
                .build());
        verify(observer).onNext(RpcPushResponseFragment.newBuilder()
                .setProgressUpdate(RpcPushProgressFragment.newBuilder()
                        .addProgressUpdates(RpcProgressUpdate.newBuilder().setPercentage(7).setMessage("message").build())
                        .build())
                .build());
        verify(observer).onCompleted();
        verifyNoMoreInteractions(observer);
    }
}