package com.atlassian.bitbucket.scm.cache.internal;

import com.atlassian.bitbucket.cluster.ClusterService;
import com.atlassian.cache.*;
import com.google.common.base.Predicates;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;

import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.isNull;
import static org.mockito.Mockito.*;

@RunWith(MockitoJUnitRunner.class)
@SuppressWarnings("unchecked")
public class DefaultClusterNotificationServiceTest {

    static final String MEMBER_1_ID = "member1";
    static final String MEMBER_2_ID = "member2";

    @Mock
    private Cache<String, String> cache;
    private CacheEntryListener<String, String> cacheListener;
    @Mock
    private CacheFactory cacheFactory;
    @Mock
    private ClusterService clusterService;
    private int counter;
    @Mock
    private NotificationListener listener;
    private DefaultClusterNotificationService service;

    @Before
    public void setup() {
        Cache<String, String> cache = mock(Cache.class);
        when(cacheFactory.<String, String>getCache(anyString(), isNull(CacheLoader.class), any(CacheSettings.class)))
                .thenReturn(cache);
        when(clusterService.isAvailable()).thenReturn(true);
        when(clusterService.getNodeId()).thenReturn(MEMBER_1_ID);

        service = new DefaultClusterNotificationService(clusterService, cacheFactory);
        ArgumentCaptor<CacheEntryListener> captor = ArgumentCaptor.forClass(CacheEntryListener.class);
        verify(cache).addListener(captor.capture(), eq(true));
        cacheListener = captor.getValue();
    }

    @Test
    public void testSubscribeAndSendFromLocalNode() {
        service.subscribe(Predicates.equalTo("test"), listener);

        // notifications from the local node should never be dispatched
        service.send("test");
        verify(listener, never()).onNotification(anyString());

        mockCacheUpdate("test", MEMBER_1_ID);
        verify(listener, never()).onNotification(anyString());
    }

    @Test
    public void testSubscribeAndSendFromOtherNode() {
        service.subscribe(Predicates.equalTo("test"), listener);

        mockCacheUpdate("test", MEMBER_2_ID);
        verify(listener).onNotification("test");
    }

    private void mockCacheUpdate(String message, String nodeId) {
        CacheEntryEvent<String, String> event = mock(CacheEntryEvent.class);
        when(event.getKey()).thenReturn(message);
        when(event.getValue()).thenReturn(nodeId + counter++);

        cacheListener.onUpdate(event);
    }
}