Unified interface

gradle.properties

@@ -1,4 +1,4 @@
-version=0.91.3
+version=0.91.4
 springVersion=4.3.3.RELEASE
 springBootVersion=1.4.1.RELEASE
 jerseyVersion=2.24

micro-couchbase/build.gradle

@@ -11,6 +11,7 @@ dependencies {
 	compile project(':micro-manifest-comparator')
 	compile project(':micro-core')
 	compile project(':micro-guava')
+	compile project(':micro-events')
 	testCompile group: 'org.codehaus.groovy', name: 'groovy-all', version:'2.3.3'
 	testCompile(group: 'org.spockframework', name: 'spock-core', version:'0.7-groovy-2.0') { exclude(module: 'groovy-all') }
 	testCompile group: 'com.cyrusinnovation', name: 'mockito-groovy-support', version:'1.3'

micro-couchbase/src/main/java/com/aol/micro/server/couchbase/ConfigureCacheTester.java

@@ -0,0 +1,21 @@
+package com.aol.micro.server.couchbase;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.Scheduled;
+
+/**
+ * Created by gordonmorrow on 03/07/2017.
+ */
+@Configuration
+public class ConfigureCacheTester {
+
+    @Autowired
+    private CouchbaseConnectionTester couchbaseConnectionTester;
+
+
+    @Scheduled(fixedDelay = 60000)
+    public synchronized void runCouchbaseConnectionTester(){
+        couchbaseConnectionTester.scheduleAndLog();
+    }
+}

micro-couchbase/src/main/java/com/aol/micro/server/couchbase/ConfigureCouchbase.java

@@ -14,7 +14,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.util.StringUtils;
 
-import com.aol.micro.server.couchbase.base.CouchbaseManifestComparator;
+import com.aol.micro.server.couchbase.manifest.comparator.CouchbaseManifestComparator;
 import com.couchbase.client.CouchbaseClient;
 import com.couchbase.client.CouchbaseConnectionFactory;
 import com.couchbase.client.CouchbaseConnectionFactoryBuilder;
@@ -45,15 +45,26 @@ public class ConfigureCouchbase {
     @Value("${couchbaseClientOperationTimeout:120000}")
     private long opTimeout;
 
+    @Value("${distributed.cache.default.expiration:691200}")
+    private int expiresAfterSeconds = 691200;
+
+    @Value("${distributed.cache.maxTry:5}")
+    private int maxTry = 5;
+
+    @Value("${distributed.cache.retryAfterSec:1}")
+    private int retryAfterSec = 1;
+
     @SuppressWarnings("rawtypes")
     @Bean(name = "couchbaseDistributedMap")
     public CouchbaseDistributedMapClient simpleCouchbaseClient() throws IOException, URISyntaxException {
         if (couchbaseClientEnabled) {
             return new CouchbaseDistributedMapClient(
-                                                     couchbaseClient());
+                                                     couchbaseClient(), expiresAfterSeconds, maxTry,
+                    retryAfterSec);
         } else {
             return new CouchbaseDistributedMapClient(
-                                                     null);
+                                                     null, expiresAfterSeconds, maxTry,
+                    retryAfterSec);
         }
     }
 

micro-couchbase/src/main/java/com/aol/micro/server/couchbase/CouchbaseConnectionTester.java

@@ -0,0 +1,57 @@
+package com.aol.micro.server.couchbase;
+
+/**
+ * Created by gordonmorrow on 03/07/2017.
+ */
+
+import com.aol.micro.server.distributed.DistributedMap;
+import com.aol.micro.server.events.ScheduledJob;
+import com.aol.micro.server.events.SystemData;
+import com.couchbase.client.CouchbaseClient;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Random;
+
+@Slf4j
+    public class CouchbaseConnectionTester implements ScheduledJob {
+
+        private static final Random random = new Random();
+
+        private final DistributedMap cache;
+        private final CouchbaseClient couchbaseClient;
+
+        public CouchbaseConnectionTester(DistributedMap cache, CouchbaseClient couchbaseClient) {
+
+            this.cache = cache;
+            this.couchbaseClient = couchbaseClient;
+        }
+
+        @Override
+        public SystemData scheduleAndLog() {
+
+            log.trace("runTestConnection()...");
+            boolean result = false;
+            try {
+                result = testConnection();
+            } catch (RuntimeException e) {
+                log.debug("Could not connect to Cache" + e.getMessage());
+            }
+                cache.setConnectionTested(result);
+
+            log.debug("Testing Couchbase connection: {}", result);
+            return null;
+
+        }
+
+        private boolean testConnection() {
+            String key = "PING_TEST";
+            log.trace("Testing connection using key {}", key);
+
+            int testValue = random.nextInt(1111);
+            couchbaseClient.set(key, 120, testValue);
+            int received = (Integer) couchbaseClient.get(key);
+
+            return received == testValue;
+        }
+
+}

micro-couchbase/src/main/java/com/aol/micro/server/couchbase/CouchbaseDistributedMapClient.java

@@ -4,31 +4,112 @@
 import java.util.concurrent.ExecutionException;
 
 import com.aol.cyclops2.util.ExceptionSoftener;
+import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.aol.micro.server.distributed.DistributedMap;
 import com.couchbase.client.CouchbaseClient;
 
+@Slf4j
 public class CouchbaseDistributedMapClient<V> implements DistributedMap<V> {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
+    private volatile boolean available = false;
 
     private final Optional<CouchbaseClient> couchbaseClient;
+    private final int expiresAfterSeconds, maxTry, retryAfterSec;
 
-    public CouchbaseDistributedMapClient(CouchbaseClient couchbaseClient) {
+    public CouchbaseDistributedMapClient(CouchbaseClient couchbaseClient, final int expiresAfterSeconds,
+                                         final int maxTry, final int retryAfterSec) {
 
         this.couchbaseClient = Optional.ofNullable(couchbaseClient);
+        this.expiresAfterSeconds = expiresAfterSeconds;
+        this.maxTry = maxTry;
+        this.retryAfterSec = retryAfterSec;
     }
 
     @Override
     public boolean put(final String key, final V value) {
+
+        log.trace("put '{}', value:{}", key, value);
+        boolean success = false;
+        int tryCount = 0;
+
+        do {
+            try {
+                if (tryCount > 0) {
+                    Thread.sleep(retryAfterSec * 1000);
+                    log.warn("retry #{}", tryCount);
+                }
+                tryCount++;
+                success = couchbaseClient.map(c -> putInternal(c, key, value))
+                        .orElse(false);
+
+            } catch (final Exception e) {
+
+                log.warn("memcache put: {}", e.getMessage());
+            }
+        } while (!success && tryCount < maxTry);
+
+        if (!success) {
+            log.error("Failed to place item in couchbase");
+        }
+        if (success && tryCount > 1) {
+            log.info("Connection restored OK");
+        }
+
+        available = success;
+
+        return success;
+    }
+
+    @Override
+    public boolean put(final String key, int expiry, final V value) {
         logger.debug("put '{}', value:{}", key, value);
-        return couchbaseClient.map(c -> putInternal(c, key, value))
-                              .orElse(false);
+        boolean success = false;
+        int tryCount = 0;
+
+        do {
+            try {
+                if (tryCount > 0) {
+                    Thread.sleep(retryAfterSec * 1000);
+                    log.warn("retry #{}", tryCount);
+                }
+                tryCount++;
+                success = couchbaseClient.map(c -> putInternalWithExpiry(c, key, value, expiry))
+                .orElse(false);
+            } catch (final Exception e) {
+
+                log.warn("memcache put: {}", e.getMessage());
+            }
+        } while (!success && tryCount < maxTry);
+
+        if (!success) {
+            log.error("Failed to place item in couchbase");
+        }
+        if (success && tryCount > 1) {
+            log.info("Connection restored OK");
+        }
+
+        available = success;
+
+        return success;
 
     }
 
+    private boolean putInternalWithExpiry(final CouchbaseClient client, final String key, final V value, int expiry) {
+
+        try {
+            return client.set(key,expiry, value)
+                    .get();
+        } catch (InterruptedException | ExecutionException e) {
+            throw ExceptionSoftener.throwSoftenedException(e);
+
+        }
+    }
+
+
     private boolean putInternal(final CouchbaseClient client, final String key, final V value) {
 
         try {
@@ -50,4 +131,5 @@ public Optional<V> get(String key) {
     public void delete(String key) {
         couchbaseClient.map(c -> c.delete(key));
     }
+
 }

micro-couchbase/src/main/java/com/aol/micro/server/couchbase/base/CouchbaseManifestComparator.java → micro-couchbase/src/main/java/com/aol/micro/server/couchbase/manifest/comparator/CouchbaseManifestComparator.java

@@ -1,4 +1,4 @@
-package com.aol.micro.server.couchbase.base;
+package com.aol.micro.server.couchbase.manifest.comparator;
 
 import java.util.Date;
 import java.util.Optional;

micro-couchbase/src/test/java/app/couchbase/distributed/map/com/aol/micro/server/CouchbaseRunnerTest.java

@@ -48,7 +48,6 @@ public void stopServer() {
     }
 
     @Test
-    @Ignore
     public void runAppAndBasicTest() throws InterruptedException, ExecutionException {
         rest.get("http://localhost:8080/simple-app/couchbase/put");
         assertThat(rest.get("http://localhost:8080/simple-app/couchbase/get"), containsString("world"));

micro-couchbase/src/test/java/com/aol/micro/server/couchbase/SimpleCouchbaseClientConnectionTest.groovy

@@ -19,7 +19,7 @@ class SimpleCouchbaseClientConnectionTest {
 	@Before
 	public void setup() {
 		client = Mockito.mock(CouchbaseClient)
-		con  = new CouchbaseDistributedMapClient(client)
+		con  = new CouchbaseDistributedMapClient(client,1,1,1)
 	}
 	@Test
 	public void testDelete() {
@@ -35,7 +35,7 @@ class SimpleCouchbaseClientConnectionTest {
 
 	@Test
 	public void testGetDistributedCacheDisabled() {
-		con  = new CouchbaseDistributedMapClient(null)
+		con  = new CouchbaseDistributedMapClient(null,1,1,1)
 		Optional result = con.get("key")
 		assertThat(result, is(Optional.empty()))
 	}

micro-couchbase/src/test/java/com/aol/micro/server/couchbase/base/ManifestComparatorKeyNotFoundExceptionTest.java → micro-couchbase/src/test/java/com/aol/micro/server/couchbase/manifest/comparator/ManifestComparatorKeyNotFoundExceptionTest.java

@@ -1,4 +1,4 @@
-package com.aol.micro.server.couchbase.base;
+package com.aol.micro.server.couchbase.manifest.comparator;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertThat;

micro-couchbase/src/test/java/com/aol/micro/server/couchbase/manifest/comparator/ManifestComparatorTest.groovy

@@ -0,0 +1 @@
+package com.aol.micro.server.couchbase.manifest.comparator;

micro-elasticache/build.gradle

@@ -8,6 +8,8 @@ dependencies {
 	compile group: 'net.spy', name: 'spymemcached', version: '2.12.3'
 	compile project(':micro-core')
 	compile project(':micro-guava')
+	compile project(':micro-events')
+	compile project(':micro-manifest-comparator')
 	testCompile group: 'org.codehaus.groovy', name: 'groovy-all', version:'2.3.3'
 	testCompile(group: 'org.spockframework', name: 'spock-core', version:'0.7-groovy-2.0') { exclude(module: 'groovy-all') }
 	testCompile group: 'com.cyrusinnovation', name: 'mockito-groovy-support', version:'1.3'

micro-elasticache/src/main/java/com/aol/micro/server/elasticache/ConfigureCacheTester.java

@@ -0,0 +1,21 @@
+package com.aol.micro.server.elasticache;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.Scheduled;
+
+/**
+ * Created by gordonmorrow on 03/07/2017.
+ */
+@Configuration
+public class ConfigureCacheTester {
+
+    @Autowired
+    private ElasticacheConnectionTester elasticacheConnectionTester;
+
+
+    @Scheduled(fixedDelay = 60000)
+    public synchronized void runElasticacheConnectionTester(){
+        elasticacheConnectionTester.scheduleAndLog();
+    }
+}

micro-elasticache/src/main/java/com/aol/micro/server/elasticache/ConfigureElasticache.java

@@ -2,28 +2,20 @@
 
 
 
+import com.aol.micro.server.distributed.DistributedMap;
 import lombok.extern.slf4j.Slf4j;
-import net.spy.memcached.*;
+import net.spy.memcached.MemcachedClient;
 
 import java.io.IOException;
 import java.net.URISyntaxException;
-import java.util.ArrayList;
-
 
-import net.spy.memcached.auth.AuthDescriptor;
-import net.spy.memcached.auth.PlainCallbackHandler;
-import net.spy.memcached.MemcachedClient;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.util.StringUtils;
 
 import java.net.InetSocketAddress;
 
-import java.util.List;
-import java.util.Optional;
-
 @Slf4j
 @Configuration
 public class ConfigureElasticache {
@@ -47,7 +39,7 @@ public ConfigureElasticache( @Value("${elasticache.hostname:null}") String hostn
 
 
     @Bean(name = "transientCache")
-    public DistributedCacheManager transientCache() throws IOException, URISyntaxException {
+    public DistributedMap transientCache() throws IOException, URISyntaxException {
         try {
             log.info("Creating Memcached Data connection for elasticache cluster: {}", hostname);
             return new TransientElasticacheDataConnection(createMemcachedClient(), retryAfterSecs, maxRetries);