Correct observer callbacks in EvCacheProvider.

ribbon-client-extensions/src/main/java/com/netflix/ribbonclientextensions/evache/EvCacheProvider.java

@@ -98,6 +98,7 @@ public class EvCacheProvider<T> implements CacheProvider<T> {
                     Future<?> future = f.getKey();
                     Subscriber subscriber = f.getValue();
                     if (subscriber.isUnsubscribed()) {
+                        future.cancel(true);
                         futureMap.remove(future);
                     } else if (future.isDone()) {
                         try {
@@ -126,6 +127,7 @@ public class EvCacheProvider<T> implements CacheProvider<T> {
             } else {
                 try {
                     subscriber.onNext(future.get());
+                    subscriber.onCompleted();
                 } catch (ExecutionException e) {
                     subscriber.onError(e.getCause());
                 }

ribbon-client-extensions/src/test/java/com/netflix/ribbonclientextensions/evache/EvCacheProviderTest.java

@@ -143,7 +143,7 @@ public class EvCacheProviderTest {
     @Test
     public void testUnsubscribedBeforeFutureCompletes() throws Exception {
         expect(evCacheImplMock.getAsynchronous("test1", transcoderMock)).andReturn(cacheFutureMock);
-
+        expect(cacheFutureMock.cancel(true)).andReturn(true);
         replayAll();
 
         EvCacheOptions options = new EvCacheOptions("testApp", "test-cache", true, 100, transcoderMock, "test{id}");
@@ -153,6 +153,8 @@ public class EvCacheProviderTest {
         Subscription subscription = cacheValue.subscribe();
         subscription.unsubscribe();
 
+        Thread.sleep(10); // So the internal thread has the chance to cancel the future
+
         verifyAll();
     }
 }
\ No newline at end of file