package com.instaclustr.cassandra.operator;

import com.google.common.cache.Cache;
import com.google.common.eventbus.EventBus;
import com.google.common.reflect.TypeToken;
import com.google.common.util.concurrent.AbstractExecutionThreadService;
import com.instaclustr.cassandra.operator.event.DataCenterEvent;
import com.instaclustr.cassandra.operator.model.DataCenter;
import com.instaclustr.cassandra.operator.model.key.DataCenterKey;
import io.kubernetes.client.ApiClient;
import io.kubernetes.client.apis.CustomObjectsApi;
import io.kubernetes.client.util.Watch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.inject.Inject;
import java.util.Set;


// TODO: merge this with WatchService (once complete)
public class DataCenterWatchService extends AbstractExecutionThreadService {
    static final Logger logger = LoggerFactory.getLogger(DataCenterWatchService.class);

    private final ApiClient apiClient;
    private final CustomObjectsApi customObjectsApi;
    private final Cache<DataCenterKey, DataCenter> dataCenterCache;
    private final DataCenterEvent.Factory dataCenterEventFactory;
    private final EventBus eventBus;

    @Inject
    DataCenterWatchService(final ApiClient apiClient,
                           final CustomObjectsApi customObjectsApi,
                           final Cache<DataCenterKey, DataCenter> dataCenterCache,
                           final DataCenterEvent.Factory dataCenterEventFactory,
                           final EventBus eventBus) {
        this.apiClient = apiClient;
        this.customObjectsApi = customObjectsApi;
        this.dataCenterCache = dataCenterCache;
        this.dataCenterEventFactory = dataCenterEventFactory;
        this.eventBus = eventBus;

    }


    private enum ResponseType {
        ADDED,
        MODIFIED,
        DELETED,
        ERROR
    }

    // TODO: make the watches configurable via injection of something like the WatchConfig object below.

//    static class WatchConfig<K, T> {
//        public final Call watchCall;
//        public final Type responseType = new TypeToken<Watch.Response<T>>() {}.getType();
//        final WatchEvent.Factory<T> watchEventFactory;
//        final Cache<K, T> responseCache;
//
//
//    }


    protected void run() throws Exception {
        String resourceVersion = null;

        while (isRunning()) {
            try {
                final Watch<DataCenter> dataCentersWatch = Watch.createWatch(apiClient,
                        customObjectsApi.listClusterCustomObjectCall("stable.instaclustr.com", "v1", "datacentres", null, null, resourceVersion, true, null, null),
                        new TypeToken<Watch.Response<DataCenter>>() {}.getType()
                );
//
                for (final Watch.Response<DataCenter> objectResponse : dataCentersWatch) {
                    final DataCenter object = objectResponse.object;

                    resourceVersion = object.getMetadata().getResourceVersion();

                    logger.debug("Watch over datacenters received {}", objectResponse.type);

                    switch (ResponseType.valueOf(objectResponse.type)) {
                        case ADDED:
                            dataCenterCache.put(DataCenterKey.forDataCenter(object), object);
                            eventBus.post(dataCenterEventFactory.createAddedEvent(object));
                            break;

                        case MODIFIED:
                        	dataCenterCache.put(DataCenterKey.forDataCenter(object), object);
                            eventBus.post(dataCenterEventFactory.createModifiedEvent(object, object));
                            break;

                        case DELETED:
                        	dataCenterCache.invalidate(DataCenterKey.forDataCenter(object));
                            eventBus.post(dataCenterEventFactory.createDeletedEvent(object));
                            break;

                        case ERROR:
                            // TODO: handle error? -- maybe?
                            break;
                    }

                }

            } catch (final RuntimeException e) {
                final Throwable cause = e.getCause();

                if (cause instanceof java.net.SocketTimeoutException)
                    continue;

                throw e; // TODO: maybe unwrap/throw `cause` -- but its a Throwable
            }
        }
    }
}

@@ -1215,39 +1215,24 @@
  apiClient,%0A
-               
             
@@ -1272,31 +1272,16 @@
 ctsApi,%0A
-               
         
@@ -1340,31 +1340,16 @@
 rCache,%0A
-               
         
@@ -1406,31 +1406,16 @@
 actory,%0A
-               
         
@@ -1893,16 +1893,20 @@
 below.%0A%0A
+    
 //    st
@@ -1936,16 +1936,20 @@
 K, T%3E %7B%0A
+    
 //      
@@ -1979,16 +1979,20 @@
 chCall;%0A
+    
 //      
@@ -2075,16 +2075,20 @@
 Type();%0A
+    
 //      
@@ -2136,16 +2136,20 @@
 actory;%0A
+    
 //      
@@ -2187,14 +2187,26 @@
 he;%0A
-//%0A//%0A
+    //%0A    //%0A    
 //  
@@ -2692,19 +2692,43 @@
         
-);%0A
+        );%0A                
 //%0A     
@@ -3148,28 +3148,24 @@
             
-    
 case ADDED:%0A
@@ -3160,28 +3160,24 @@
 case ADDED:%0A
-    
             
@@ -3270,36 +3270,32 @@
                 
-    
 eventBus.post(da
@@ -3362,36 +3362,32 @@
             
-  
-  
 break;%0A%0A
             
@@ -3366,36 +3366,32 @@
         break;%0A%0A
-    
                 
@@ -3425,33 +3425,32 @@
                 
-%09
 dataCenterCache.
@@ -3515,36 +3515,32 @@
                 
-    
 eventBus.post(da
@@ -3618,36 +3618,32 @@
             
-  
-  
 break;%0A%0A
             
@@ -3622,36 +3622,32 @@
         break;%0A%0A
-    
                 
@@ -3692,9 +3692,8 @@
     
-%09
 data
@@ -3769,36 +3769,32 @@
                 
-    
 eventBus.post(da
@@ -3863,36 +3863,32 @@
             
-   
- 
 break;%0A%0A
             
@@ -3867,36 +3867,32 @@
         break;%0A%0A
-    
                 
@@ -3907,20 +3907,16 @@
  ERROR:%0A
-    
         
@@ -3964,20 +3964,16 @@
  maybe?%0A
-    
         

correct indentation