killbill-uncached

diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/api/DefaultEntitlement.java b/entitlement/src/main/java/com/ning/billing/entitlement/api/DefaultEntitlement.java
index 85d1a1c..983381d 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/api/DefaultEntitlement.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/api/DefaultEntitlement.java
@@ -233,7 +233,6 @@ public class DefaultEntitlement extends EntityBase implements Entitlement {
         return getSubscriptionBase().getLastActiveCategory();
     }
 
-
     @Override
     public Entitlement cancelEntitlementWithPolicy(final EntitlementActionPolicy entitlementPolicy, final CallContext callContext) throws EntitlementApiException {
         // Get the latest state from disk - required to have the latest CTD
@@ -461,7 +460,7 @@ public class DefaultEntitlement extends EntityBase implements Entitlement {
             // Note that usually we record the notification from the DAO. We cannot do it here because not all calls
             // go through the DAO (e.g. change)
             final boolean isBaseEntitlementCancelled = eventsStream.isEntitlementCancelled();
-            final NotificationEvent notificationEvent = new EntitlementNotificationKey(getId(), isBaseEntitlementCancelled ? EntitlementNotificationKeyAction.CANCEL : EntitlementNotificationKeyAction.CHANGE, effectiveDate);
+            final NotificationEvent notificationEvent = new EntitlementNotificationKey(getId(), getBundleId(), isBaseEntitlementCancelled ? EntitlementNotificationKeyAction.CANCEL : EntitlementNotificationKeyAction.CHANGE, effectiveDate);
             recordFutureNotification(effectiveDate, notificationEvent, internalCallContext);
             return;
         }

diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/api/DefaultEntitlementApi.java b/entitlement/src/main/java/com/ning/billing/entitlement/api/DefaultEntitlementApi.java
index 98c7a8b..5be1c0e 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/api/DefaultEntitlementApi.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/api/DefaultEntitlementApi.java
@@ -16,6 +16,7 @@
 
 package com.ning.billing.entitlement.api;
 
+import java.io.IOException;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -40,15 +41,21 @@ import com.ning.billing.callcontext.InternalTenantContext;
 import com.ning.billing.catalog.api.BillingActionPolicy;
 import com.ning.billing.catalog.api.PlanPhaseSpecifier;
 import com.ning.billing.clock.Clock;
+import com.ning.billing.entitlement.DefaultEntitlementService;
 import com.ning.billing.entitlement.EntitlementService;
 import com.ning.billing.entitlement.EntitlementTransitionType;
 import com.ning.billing.entitlement.block.BlockingChecker;
 import com.ning.billing.entitlement.dao.BlockingStateDao;
+import com.ning.billing.entitlement.engine.core.EntitlementNotificationKey;
+import com.ning.billing.entitlement.engine.core.EntitlementNotificationKeyAction;
 import com.ning.billing.entitlement.engine.core.EntitlementUtils;
 import com.ning.billing.entitlement.engine.core.EventsStream;
 import com.ning.billing.entitlement.engine.core.EventsStreamBuilder;
 import com.ning.billing.junction.DefaultBlockingState;
+import com.ning.billing.notificationq.api.NotificationEvent;
+import com.ning.billing.notificationq.api.NotificationQueue;
 import com.ning.billing.notificationq.api.NotificationQueueService;
+import com.ning.billing.notificationq.api.NotificationQueueService.NoSuchNotificationQueue;
 import com.ning.billing.subscription.api.SubscriptionBase;
 import com.ning.billing.subscription.api.SubscriptionBaseInternalApi;
 import com.ning.billing.subscription.api.transfer.SubscriptionBaseTransferApi;
@@ -84,7 +91,7 @@ public class DefaultEntitlementApi implements EntitlementApi {
     private final PersistentBus eventBus;
     private final EventsStreamBuilder eventsStreamBuilder;
     private final EntitlementUtils entitlementUtils;
-    protected final NotificationQueueService notificationQueueService;
+    private final NotificationQueueService notificationQueueService;
 
     @Inject
     public DefaultEntitlementApi(final PersistentBus eventBus, final InternalCallContextFactory internalCallContextFactory,
@@ -234,9 +241,9 @@ public class DefaultEntitlementApi implements EntitlementApi {
             final SubscriptionBase baseSubscription = subscriptionInternalApi.getBaseSubscription(bundleId, contextWithValidAccountRecordId);
             final DateTime effectiveDate = dateHelper.fromLocalDateAndReferenceTime(localEffectiveDate, baseSubscription.getStartDate(), contextWithValidAccountRecordId);
 
-            // STEPH TODO implement ability to pause in the future
             if (!dateHelper.isBeforeOrEqualsToday(effectiveDate, account.getTimeZone())) {
-                throw new UnsupportedOperationException("Pausing with a future date has not been implemented yet");
+                recordPauseResumeNotificationEntry(baseSubscription.getId(), bundleId, effectiveDate, true, contextWithValidAccountRecordId);
+                return;
             }
 
             final BlockingState state = new DefaultBlockingState(bundleId, BlockingStateType.SUBSCRIPTION_BUNDLE, ENT_STATE_BLOCKED, EntitlementService.ENTITLEMENT_SERVICE_NAME, true, true, true, effectiveDate);
@@ -266,20 +273,22 @@ public class DefaultEntitlementApi implements EntitlementApi {
     public void resume(final UUID bundleId, final LocalDate localEffectiveDate, final CallContext context) throws EntitlementApiException {
         try {
             final InternalCallContext contextWithValidAccountRecordId = internalCallContextFactory.createInternalCallContext(bundleId, ObjectType.BUNDLE, context);
-            final BlockingState currentState = blockingStateDao.getBlockingStateForService(bundleId, BlockingStateType.SUBSCRIPTION_BUNDLE, EntitlementService.ENTITLEMENT_SERVICE_NAME, contextWithValidAccountRecordId);
-            if (currentState == null || currentState.getStateName().equals(ENT_STATE_CLEAR)) {
-                // Nothing to do.
-                return;
-            }
             final SubscriptionBaseBundle bundle = subscriptionInternalApi.getBundleFromId(bundleId, contextWithValidAccountRecordId);
             final Account account = accountApi.getAccountById(bundle.getAccountId(), contextWithValidAccountRecordId);
             final SubscriptionBase baseSubscription = subscriptionInternalApi.getBaseSubscription(bundleId, contextWithValidAccountRecordId);
 
             final DateTime effectiveDate = dateHelper.fromLocalDateAndReferenceTime(localEffectiveDate, baseSubscription.getStartDate(), contextWithValidAccountRecordId);
 
-            // STEPH TODO implement ability to pause in the future
             if (!dateHelper.isBeforeOrEqualsToday(effectiveDate, account.getTimeZone())) {
-                throw new UnsupportedOperationException("Resuming with a future date has not been implemented yet");
+                recordPauseResumeNotificationEntry(baseSubscription.getId(), bundleId, effectiveDate, false, contextWithValidAccountRecordId);
+                return;
+            }
+
+            final BlockingState currentState = blockingStateDao.getBlockingStateForService(bundleId, BlockingStateType.SUBSCRIPTION_BUNDLE, EntitlementService.ENTITLEMENT_SERVICE_NAME, contextWithValidAccountRecordId);
+            if (currentState == null || currentState.getStateName().equals(ENT_STATE_CLEAR)) {
+                // Nothing to do.
+                log.warn("Current state is {}, nothing to resume", currentState);
+                return;
             }
 
             final BlockingState state = new DefaultBlockingState(bundleId, BlockingStateType.SUBSCRIPTION_BUNDLE, ENT_STATE_CLEAR, EntitlementService.ENTITLEMENT_SERVICE_NAME, false, false, false, effectiveDate);
@@ -349,4 +358,21 @@ public class DefaultEntitlementApi implements EntitlementApi {
             throw new EntitlementApiException(e);
         }
     }
+
+    private void recordPauseResumeNotificationEntry(final UUID entitlementId, final UUID bundleId, final DateTime effectiveDate, final boolean isPause, final InternalCallContext contextWithValidAccountRecordId) throws EntitlementApiException {
+        final NotificationEvent notificationEvent = new EntitlementNotificationKey(entitlementId,
+                                                                                   bundleId,
+                                                                                   isPause ? EntitlementNotificationKeyAction.PAUSE : EntitlementNotificationKeyAction.RESUME,
+                                                                                   effectiveDate);
+
+        try {
+            final NotificationQueue subscriptionEventQueue = notificationQueueService.getNotificationQueue(DefaultEntitlementService.ENTITLEMENT_SERVICE_NAME,
+                                                                                                           DefaultEntitlementService.NOTIFICATION_QUEUE_NAME);
+            subscriptionEventQueue.recordFutureNotification(effectiveDate, notificationEvent, contextWithValidAccountRecordId.getUserToken(), contextWithValidAccountRecordId.getAccountRecordId(), contextWithValidAccountRecordId.getTenantRecordId());
+        } catch (final NoSuchNotificationQueue e) {
+            throw new EntitlementApiException(e, ErrorCode.__UNKNOWN_ERROR_CODE);
+        } catch (final IOException e) {
+            throw new EntitlementApiException(e, ErrorCode.__UNKNOWN_ERROR_CODE);
+        }
+    }
 }

diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/DefaultEntitlementService.java b/entitlement/src/main/java/com/ning/billing/entitlement/DefaultEntitlementService.java
index 7e7174a..e6ff227 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/DefaultEntitlementService.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/DefaultEntitlementService.java
@@ -130,13 +130,17 @@ public class DefaultEntitlementService implements EntitlementService {
         }
 
         final EntitlementNotificationKeyAction entitlementNotificationKeyAction = key.getEntitlementNotificationKeyAction();
-        if (EntitlementNotificationKeyAction.CHANGE.equals(entitlementNotificationKeyAction) ||
-            EntitlementNotificationKeyAction.CANCEL.equals(entitlementNotificationKeyAction)) {
-            try {
+        try {
+            if (EntitlementNotificationKeyAction.CHANGE.equals(entitlementNotificationKeyAction) ||
+                EntitlementNotificationKeyAction.CANCEL.equals(entitlementNotificationKeyAction)) {
                 ((DefaultEntitlement) entitlement).blockAddOnsIfRequired(key.getEffectiveDate(), internalCallContext.toTenantContext(tenantId), internalCallContext);
-            } catch (EntitlementApiException e) {
-                log.error("Error processing event for entitlement {}" + entitlement.getId(), e);
+            } else if (EntitlementNotificationKeyAction.PAUSE.equals(entitlementNotificationKeyAction)) {
+                entitlementApi.pause(key.getBundleId(), key.getEffectiveDate().toLocalDate(), internalCallContext.toCallContext());
+            } else if (EntitlementNotificationKeyAction.RESUME.equals(entitlementNotificationKeyAction)) {
+                entitlementApi.resume(key.getBundleId(), key.getEffectiveDate().toLocalDate(), internalCallContext.toCallContext());
             }
+        } catch (final EntitlementApiException e) {
+            log.error("Error processing event for entitlement {}" + entitlement.getId(), e);
         }
     }
 

diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/engine/core/EntitlementNotificationKey.java b/entitlement/src/main/java/com/ning/billing/entitlement/engine/core/EntitlementNotificationKey.java
index b9da6a6..c3c93e2 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/engine/core/EntitlementNotificationKey.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/engine/core/EntitlementNotificationKey.java
@@ -28,14 +28,17 @@ import com.fasterxml.jackson.annotation.JsonProperty;
 public class EntitlementNotificationKey implements NotificationEvent {
 
     private final UUID entitlementId;
+    private final UUID bundleId;
     private final EntitlementNotificationKeyAction entitlementNotificationKeyAction;
     private final DateTime effectiveDate;
 
     @JsonCreator
     public EntitlementNotificationKey(@JsonProperty("entitlementId") final UUID entitlementId,
+                                      @JsonProperty("bundleId") final UUID bundleId,
                                       @JsonProperty("entitlementNotificationKeyAction") final EntitlementNotificationKeyAction entitlementNotificationKeyAction,
                                       @JsonProperty("effectiveDate") final DateTime effectiveDate) {
         this.entitlementId = entitlementId;
+        this.bundleId = bundleId;
         this.entitlementNotificationKeyAction = entitlementNotificationKeyAction;
         this.effectiveDate = effectiveDate;
     }
@@ -44,6 +47,10 @@ public class EntitlementNotificationKey implements NotificationEvent {
         return entitlementId;
     }
 
+    public UUID getBundleId() {
+        return bundleId;
+    }
+
     public EntitlementNotificationKeyAction getEntitlementNotificationKeyAction() {
         return entitlementNotificationKeyAction;
     }
@@ -56,6 +63,7 @@ public class EntitlementNotificationKey implements NotificationEvent {
     public String toString() {
         final StringBuilder sb = new StringBuilder("EntitlementNotificationKey{");
         sb.append("entitlementId=").append(entitlementId);
+        sb.append(", bundleId=").append(bundleId);
         sb.append(", entitlementNotificationKeyAction=").append(entitlementNotificationKeyAction);
         sb.append(", effectiveDate=").append(effectiveDate);
         sb.append('}');
@@ -76,6 +84,9 @@ public class EntitlementNotificationKey implements NotificationEvent {
         if (entitlementId != null ? !entitlementId.equals(that.entitlementId) : that.entitlementId != null) {
             return false;
         }
+        if (bundleId != null ? !bundleId.equals(that.bundleId) : that.bundleId != null) {
+            return false;
+        }
         if (entitlementNotificationKeyAction != that.entitlementNotificationKeyAction) {
             return false;
         }
@@ -89,6 +100,7 @@ public class EntitlementNotificationKey implements NotificationEvent {
     @Override
     public int hashCode() {
         int result = entitlementId != null ? entitlementId.hashCode() : 0;
+        result = 31 * result + (bundleId != null ? bundleId.hashCode() : 0);
         result = 31 * result + (entitlementNotificationKeyAction != null ? entitlementNotificationKeyAction.hashCode() : 0);
         result = 31 * result + (effectiveDate != null ? effectiveDate.hashCode() : 0);
         return result;

diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/engine/core/EntitlementNotificationKeyAction.java b/entitlement/src/main/java/com/ning/billing/entitlement/engine/core/EntitlementNotificationKeyAction.java
index 446cfc9..487854c 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/engine/core/EntitlementNotificationKeyAction.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/engine/core/EntitlementNotificationKeyAction.java
@@ -18,5 +18,7 @@ package com.ning.billing.entitlement.engine.core;
 
 public enum EntitlementNotificationKeyAction {
     CANCEL,
-    CHANGE
+    CHANGE,
+    PAUSE,
+    RESUME
 }

diff --git a/entitlement/src/test/java/com/ning/billing/entitlement/api/TestDefaultEntitlementApi.java b/entitlement/src/test/java/com/ning/billing/entitlement/api/TestDefaultEntitlementApi.java
index a27d833..e2ff2d8 100644
--- a/entitlement/src/test/java/com/ning/billing/entitlement/api/TestDefaultEntitlementApi.java
+++ b/entitlement/src/test/java/com/ning/billing/entitlement/api/TestDefaultEntitlementApi.java
@@ -40,7 +40,6 @@ import com.ning.billing.entitlement.api.Entitlement.EntitlementState;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNull;
 
-
 public class TestDefaultEntitlementApi extends EntitlementTestSuiteWithEmbeddedDB {
 
     @Test(groups = "slow")
@@ -283,7 +282,6 @@ public class TestDefaultEntitlementApi extends EntitlementTestSuiteWithEmbeddedD
             assertEquals(e.getCode(), ErrorCode.ENT_ALREADY_BLOCKED.getCode());
         }
 
-
         final List<Entitlement> bundleEntitlements2 = entitlementApi.getAllEntitlementsForBundle(telescopicEntitlement2.getBundleId(), callContext);
         assertEquals(bundleEntitlements2.size(), 2);
         for (final Entitlement cur : bundleEntitlements2) {
@@ -321,6 +319,51 @@ public class TestDefaultEntitlementApi extends EntitlementTestSuiteWithEmbeddedD
         }
     }
 
+    @Test(groups = "slow", description = "Test pause / unpause in the future")
+    public void testPauseUnpauseInTheFuture() throws AccountApiException, EntitlementApiException {
+        final LocalDate initialDate = new LocalDate(2013, 8, 7);
+        clock.setDay(initialDate);
+
+        final Account account = accountApi.createAccount(getAccountData(7), callContext);
+
+        // Create entitlement
+        testListener.pushExpectedEvent(NextEvent.CREATE);
+        final PlanPhaseSpecifier spec = new PlanPhaseSpecifier("Shotgun", ProductCategory.BASE, BillingPeriod.ANNUAL, PriceListSet.DEFAULT_PRICELIST_NAME, null);
+        final Entitlement baseEntitlement = entitlementApi.createBaseEntitlement(account.getId(), spec, account.getExternalKey(), initialDate, callContext);
+        assertListenerStatus();
+
+        // Get the phase event out of the way
+        testListener.pushExpectedEvents(NextEvent.PHASE);
+        clock.setDay(new LocalDate(2013, 9, 7));
+        assertListenerStatus();
+
+        final LocalDate pauseDate = new LocalDate(2013, 9, 17);
+        entitlementApi.pause(baseEntitlement.getBundleId(), pauseDate, callContext);
+        // No event yet
+        assertListenerStatus();
+
+        final LocalDate resumeDate = new LocalDate(2013, 12, 24);
+        entitlementApi.resume(baseEntitlement.getBundleId(), resumeDate, callContext);
+        // No event yet
+        assertListenerStatus();
+
+        testListener.pushExpectedEvents(NextEvent.PAUSE, NextEvent.BLOCK);
+        clock.setDay(pauseDate.plusDays(1));
+        assertListenerStatus();
+
+        // Verify blocking state
+        final Entitlement baseEntitlementPaused = entitlementApi.getEntitlementForId(baseEntitlement.getId(), callContext);
+        assertEquals(baseEntitlementPaused.getState(), EntitlementState.BLOCKED);
+
+        testListener.pushExpectedEvents(NextEvent.RESUME, NextEvent.BLOCK);
+        clock.setDay(resumeDate.plusDays(1));
+        assertListenerStatus();
+
+        // Verify blocking state
+        final Entitlement baseEntitlementUnpaused = entitlementApi.getEntitlementForId(baseEntitlement.getId(), callContext);
+        assertEquals(baseEntitlementUnpaused.getState(), EntitlementState.ACTIVE);
+    }
+
     @Test(groups = "slow")
     public void testTransferBundle() throws AccountApiException, EntitlementApiException {
         final LocalDate initialDate = new LocalDate(2013, 8, 7);