killbill-memoizeit
Changes
entitlement/src/main/java/com/ning/billing/entitlement/api/timeline/SubscriptionDataRepair.java 2(+1 -1)
entitlement/src/main/java/com/ning/billing/entitlement/api/user/DefaultRequestedSubscriptionEvent.java 21(+21 -0)
entitlement/src/main/java/com/ning/billing/entitlement/api/user/DefaultSubscriptionApiService.java 8(+4 -4)
entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/AuditedEntitlementDao.java 138(+111 -27)
entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/RepairEntitlementDao.java 52(+20 -32)
Details
diff --git a/analytics/src/test/java/com/ning/billing/analytics/TestBusinessTagRecorder.java b/analytics/src/test/java/com/ning/billing/analytics/TestBusinessTagRecorder.java
index 6f18739..79e762f 100644
--- a/analytics/src/test/java/com/ning/billing/analytics/TestBusinessTagRecorder.java
+++ b/analytics/src/test/java/com/ning/billing/analytics/TestBusinessTagRecorder.java
@@ -85,7 +85,7 @@ public class TestBusinessTagRecorder extends TestWithEmbeddedDB {
final CatalogService catalogService = new DefaultCatalogService(Mockito.mock(CatalogConfig.class), Mockito.mock(VersionedCatalogLoader.class));
final AddonUtils addonUtils = new AddonUtils(catalogService);
final DefaultNotificationQueueService notificationQueueService = new DefaultNotificationQueueService(dbi, clock);
- final EntitlementDao entitlementDao = new AuditedEntitlementDao(dbi, clock, addonUtils, notificationQueueService, eventBus);
+ final EntitlementDao entitlementDao = new AuditedEntitlementDao(dbi, clock, addonUtils, notificationQueueService, eventBus, catalogService);
final PlanAligner planAligner = new PlanAligner(catalogService);
final DefaultSubscriptionApiService apiService = new DefaultSubscriptionApiService(clock, entitlementDao, catalogService, planAligner);
final DefaultSubscriptionFactory subscriptionFactory = new DefaultSubscriptionFactory(apiService, clock, catalogService);
diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/api/timeline/SubscriptionDataRepair.java b/entitlement/src/main/java/com/ning/billing/entitlement/api/timeline/SubscriptionDataRepair.java
index cdc6797..66cf27c 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/api/timeline/SubscriptionDataRepair.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/api/timeline/SubscriptionDataRepair.java
@@ -165,7 +165,7 @@ public class SubscriptionDataRepair extends SubscriptionData {
.setRequestedDate(now)
.setUserToken(context.getUserToken())
.setFromDisk(true));
- repairDao.cancelSubscription(cur.getId(), cancelEvent, context, 0);
+ repairDao.cancelSubscription(cur, cancelEvent, context, 0);
cur.rebuildTransitions(repairDao.getEventsForSubscription(cur.getId()), catalogService.getFullCatalog());
}
}
diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/api/user/DefaultRequestedSubscriptionEvent.java b/entitlement/src/main/java/com/ning/billing/entitlement/api/user/DefaultRequestedSubscriptionEvent.java
index f03ba35..b9742e6 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/api/user/DefaultRequestedSubscriptionEvent.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/api/user/DefaultRequestedSubscriptionEvent.java
@@ -52,4 +52,25 @@ public class DefaultRequestedSubscriptionEvent extends DefaultSubscriptionEvent
previousPhase, previousPriceList, nextState, nextPlan, nextPhase, nextPriceList, totalOrdering, userToken,
transitionType, remainingEventsForUserOperation, startDate);
}
+
+ public DefaultRequestedSubscriptionEvent(final SubscriptionEvent event) {
+ this(event.getId(),
+ event.getSubscriptionId(),
+ event.getBundleId(),
+ event.getRequestedTransitionTime(),
+ event.getEffectiveTransitionTime(),
+ event.getPreviousState(),
+ event.getPreviousPlan(),
+ event.getPreviousPhase(),
+ event.getPreviousPriceList(),
+ event.getNextState(),
+ event.getNextPlan(),
+ event.getNextPhase(),
+ event.getNextPriceList(),
+ event.getTotalOrdering(),
+ event.getUserToken(),
+ event.getTransitionType(),
+ event.getRemainingEventsForUserOperation(),
+ event.getSubscriptionStartDate());
+ }
}
diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/api/user/DefaultSubscriptionApiService.java b/entitlement/src/main/java/com/ning/billing/entitlement/api/user/DefaultSubscriptionApiService.java
index 4469661..99e3e58 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/api/user/DefaultSubscriptionApiService.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/api/user/DefaultSubscriptionApiService.java
@@ -139,7 +139,7 @@ public class DefaultSubscriptionApiService implements SubscriptionApiService {
events.add(nextPhaseEvent);
}
if (reCreate) {
- dao.recreateSubscription(subscription.getId(), events, context);
+ dao.recreateSubscription(subscription, events, context);
} else {
dao.createSubscription(subscription, events, context);
}
@@ -179,7 +179,7 @@ public class DefaultSubscriptionApiService implements SubscriptionApiService {
.setUserToken(context.getUserToken())
.setFromDisk(true));
- dao.cancelSubscription(subscription.getId(), cancelEvent, context, 0);
+ dao.cancelSubscription(subscription, cancelEvent, context, 0);
subscription.rebuildTransitions(dao.getEventsForSubscription(subscription.getId()), catalogService.getFullCatalog());
return (policy == ActionPolicy.IMMEDIATE);
} catch (CatalogApiException e) {
@@ -213,7 +213,7 @@ public class DefaultSubscriptionApiService implements SubscriptionApiService {
uncancelEvents.add(nextPhaseEvent);
}
- dao.uncancelSubscription(subscription.getId(), uncancelEvents, context);
+ dao.uncancelSubscription(subscription, uncancelEvents, context);
subscription.rebuildTransitions(dao.getEventsForSubscription(subscription.getId()), catalogService.getFullCatalog());
return true;
@@ -285,7 +285,7 @@ public class DefaultSubscriptionApiService implements SubscriptionApiService {
changeEvents.add(nextPhaseEvent);
}
changeEvents.add(changeEvent);
- dao.changePlan(subscription.getId(), changeEvents, context);
+ dao.changePlan(subscription, changeEvents, context);
subscription.rebuildTransitions(dao.getEventsForSubscription(subscription.getId()), catalogService.getFullCatalog());
return (policy == ActionPolicy.IMMEDIATE);
} catch (CatalogApiException e) {
diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/engine/core/Engine.java b/entitlement/src/main/java/com/ning/billing/entitlement/engine/core/Engine.java
index a4a926e..1d2958e 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/engine/core/Engine.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/engine/core/Engine.java
@@ -210,10 +210,10 @@ public class Engine implements EventListener, EntitlementService {
final TimedPhase nextTimedPhase = planAligner.getNextTimedPhase(subscription, now, now);
final PhaseEvent nextPhaseEvent = (nextTimedPhase != null) ?
PhaseEventData.createNextPhaseEvent(nextTimedPhase.getPhase().getName(), subscription, now, nextTimedPhase.getStartPhase()) :
- null;
- if (nextPhaseEvent != null) {
- dao.createNextPhaseEvent(subscription.getId(), nextPhaseEvent, context);
- }
+ null;
+ if (nextPhaseEvent != null) {
+ dao.createNextPhaseEvent(subscription, nextPhaseEvent, context);
+ }
} catch (EntitlementError e) {
log.error(String.format("Failed to insert next phase for subscription %s", subscription.getId()), e);
}
@@ -229,7 +229,8 @@ public class Engine implements EventListener, EntitlementService {
final List<Subscription> subscriptions = dao.getSubscriptions(subscriptionFactory, baseSubscription.getBundleId());
- final Map<UUID, EntitlementEvent> addOnCancellations = new HashMap<UUID, EntitlementEvent>();
+ final Map<UUID, EntitlementEvent> addOnCancellationEvents = new HashMap<UUID, EntitlementEvent>();
+ final Map<UUID, SubscriptionData> addOnCancellationSubscriptions = new HashMap<UUID, SubscriptionData>();
final Iterator<Subscription> it = subscriptions.iterator();
while (it.hasNext()) {
@@ -254,13 +255,14 @@ public class Engine implements EventListener, EntitlementService {
.setUserToken(context.getUserToken())
.setFromDisk(true));
- addOnCancellations.put(cur.getId(), cancelEvent);
+ addOnCancellationEvents.put(cur.getId(), cancelEvent);
+ addOnCancellationSubscriptions.put(cur.getId(), cur);
}
}
- final int addOnSize = addOnCancellations.size();
+ final int addOnSize = addOnCancellationEvents.size();
int cancelSeq = addOnSize - 1;
- for (final UUID key : addOnCancellations.keySet()) {
- dao.cancelSubscription(key, addOnCancellations.get(key), context, cancelSeq);
+ for (final UUID key : addOnCancellationEvents.keySet()) {
+ dao.cancelSubscription(addOnCancellationSubscriptions.get(key), addOnCancellationEvents.get(key), context, cancelSeq);
cancelSeq--;
}
return addOnSize;
diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/AuditedEntitlementDao.java b/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/AuditedEntitlementDao.java
index 2d7f212..5a25c92 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/AuditedEntitlementDao.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/AuditedEntitlementDao.java
@@ -38,8 +38,10 @@ import org.slf4j.LoggerFactory;
import com.google.common.base.Predicate;
import com.google.common.collect.Collections2;
+import com.google.common.collect.ImmutableList;
import com.google.inject.Inject;
import com.ning.billing.ErrorCode;
+import com.ning.billing.catalog.api.CatalogService;
import com.ning.billing.catalog.api.Plan;
import com.ning.billing.catalog.api.ProductCategory;
import com.ning.billing.entitlement.api.SubscriptionFactory;
@@ -49,7 +51,10 @@ import com.ning.billing.entitlement.api.migration.AccountMigrationData.Subscript
import com.ning.billing.entitlement.api.timeline.DefaultRepairEntitlementEvent;
import com.ning.billing.entitlement.api.timeline.RepairEntitlementEvent;
import com.ning.billing.entitlement.api.timeline.SubscriptionDataRepair;
+import com.ning.billing.entitlement.api.user.DefaultRequestedSubscriptionEvent;
import com.ning.billing.entitlement.api.user.DefaultSubscriptionFactory.SubscriptionBuilder;
+import com.ning.billing.entitlement.api.user.EffectiveSubscriptionEvent;
+import com.ning.billing.entitlement.api.user.RequestedSubscriptionEvent;
import com.ning.billing.entitlement.api.user.Subscription;
import com.ning.billing.entitlement.api.user.SubscriptionBundle;
import com.ning.billing.entitlement.api.user.SubscriptionBundleData;
@@ -72,7 +77,6 @@ import com.ning.billing.util.callcontext.CallContext;
import com.ning.billing.util.clock.Clock;
import com.ning.billing.util.dao.EntityAudit;
import com.ning.billing.util.dao.TableName;
-import com.ning.billing.util.notificationq.NotificationKey;
import com.ning.billing.util.notificationq.NotificationQueue;
import com.ning.billing.util.notificationq.NotificationQueueService;
import com.ning.billing.util.notificationq.NotificationQueueService.NoSuchNotificationQueue;
@@ -87,10 +91,11 @@ public class AuditedEntitlementDao implements EntitlementDao {
private final NotificationQueueService notificationQueueService;
private final AddonUtils addonUtils;
private final Bus eventBus;
+ private final CatalogService catalogService;
@Inject
- public AuditedEntitlementDao(final IDBI dbi, final Clock clock, final AddonUtils addonUtils,
- final NotificationQueueService notificationQueueService, final Bus eventBus) {
+ public AuditedEntitlementDao(final IDBI dbi, final Clock clock, final AddonUtils addonUtils, final NotificationQueueService notificationQueueService,
+ final Bus eventBus, final CatalogService catalogService) {
this.clock = clock;
this.subscriptionsDao = dbi.onDemand(SubscriptionSqlDao.class);
this.eventsDao = dbi.onDemand(EntitlementEventSqlDao.class);
@@ -98,6 +103,7 @@ public class AuditedEntitlementDao implements EntitlementDao {
this.notificationQueueService = notificationQueueService;
this.addonUtils = addonUtils;
this.eventBus = eventBus;
+ this.catalogService = catalogService;
}
@Override
@@ -206,10 +212,13 @@ public class AuditedEntitlementDao implements EntitlementDao {
}
@Override
- public void createNextPhaseEvent(final UUID subscriptionId, final EntitlementEvent nextPhase, final CallContext context) {
+ public void createNextPhaseEvent(final SubscriptionData subscription, final EntitlementEvent nextPhase, final CallContext context) {
eventsDao.inTransaction(new Transaction<Void, EntitlementEventSqlDao>() {
@Override
public Void inTransaction(final EntitlementEventSqlDao transactional, final TransactionStatus status) throws Exception {
+ final UUID subscriptionId = subscription.getId();
+ final EntitlementEvent prevPhase = getLastStoredEventForSubscription(subscriptionId);
+
cancelNextPhaseEventFromTransaction(subscriptionId, transactional, context);
transactional.insertEvent(nextPhase, context);
@@ -218,7 +227,9 @@ public class AuditedEntitlementDao implements EntitlementDao {
transactional.insertAuditFromTransaction(audit, context);
recordFutureNotificationFromTransaction(transactional,
- nextPhase.getEffectiveDate(),
+ subscription,
+ prevPhase,
+ nextPhase,
new EntitlementNotificationKey(nextPhase.getId()));
return null;
@@ -280,13 +291,19 @@ public class AuditedEntitlementDao implements EntitlementDao {
final EntitlementEventSqlDao eventsDaoFromSameTransaction = transactional.become(EntitlementEventSqlDao.class);
final List<EntityAudit> audits = new ArrayList<EntityAudit>();
+ EntitlementEvent prev = null;
for (final EntitlementEvent cur : initialEvents) {
eventsDaoFromSameTransaction.insertEvent(cur, context);
+
final Long recordId = eventsDaoFromSameTransaction.getRecordId(cur.getId().toString());
audits.add(new EntityAudit(TableName.SUBSCRIPTION_EVENTS, recordId, ChangeType.INSERT));
+
recordFutureNotificationFromTransaction(transactional,
- cur.getEffectiveDate(),
+ subscription,
+ prev,
+ cur,
new EntitlementNotificationKey(cur.getId()));
+ prev = cur;
}
eventsDaoFromSameTransaction.insertAuditFromTransaction(audits, context);
@@ -296,20 +313,22 @@ public class AuditedEntitlementDao implements EntitlementDao {
}
@Override
- public void recreateSubscription(final UUID subscriptionId, final List<EntitlementEvent> recreateEvents, final CallContext context) {
+ public void recreateSubscription(final SubscriptionData subscription, final List<EntitlementEvent> recreateEvents, final CallContext context) {
eventsDao.inTransaction(new Transaction<Void, EntitlementEventSqlDao>() {
@Override
- public Void inTransaction(final EntitlementEventSqlDao transactional,
- final TransactionStatus status) throws Exception {
+ public Void inTransaction(final EntitlementEventSqlDao transactional, final TransactionStatus status) throws Exception {
final List<EntityAudit> audits = new ArrayList<EntityAudit>();
+ EntitlementEvent prev = null;
for (final EntitlementEvent cur : recreateEvents) {
transactional.insertEvent(cur, context);
final Long recordId = transactional.getRecordId(cur.getId().toString());
audits.add(new EntityAudit(TableName.SUBSCRIPTION_EVENTS, recordId, ChangeType.INSERT));
recordFutureNotificationFromTransaction(transactional,
- cur.getEffectiveDate(),
+ subscription,
+ prev,
+ cur,
new EntitlementNotificationKey(cur.getId()));
-
+ prev = cur;
}
transactional.insertAuditFromTransaction(audits, context);
@@ -319,10 +338,13 @@ public class AuditedEntitlementDao implements EntitlementDao {
}
@Override
- public void cancelSubscription(final UUID subscriptionId, final EntitlementEvent cancelEvent, final CallContext context, final int seqId) {
+ public void cancelSubscription(final SubscriptionData subscription, final EntitlementEvent cancelEvent, final CallContext context, final int seqId) {
eventsDao.inTransaction(new Transaction<Void, EntitlementEventSqlDao>() {
@Override
public Void inTransaction(final EntitlementEventSqlDao transactional, final TransactionStatus status) throws Exception {
+ final UUID subscriptionId = subscription.getId();
+ final EntitlementEvent prevEvent = getLastStoredEventForSubscription(subscriptionId);
+
cancelNextCancelEventFromTransaction(subscriptionId, transactional, context);
cancelNextChangeEventFromTransaction(subscriptionId, transactional, context);
cancelNextPhaseEventFromTransaction(subscriptionId, transactional, context);
@@ -334,7 +356,9 @@ public class AuditedEntitlementDao implements EntitlementDao {
transactional.insertAuditFromTransaction(audit, context);
recordFutureNotificationFromTransaction(transactional,
- cancelEvent.getEffectiveDate(),
+ subscription,
+ prevEvent,
+ cancelEvent,
new EntitlementNotificationKey(cancelEvent.getId(), seqId));
return null;
}
@@ -342,10 +366,11 @@ public class AuditedEntitlementDao implements EntitlementDao {
}
@Override
- public void uncancelSubscription(final UUID subscriptionId, final List<EntitlementEvent> uncancelEvents, final CallContext context) {
+ public void uncancelSubscription(final SubscriptionData subscription, final List<EntitlementEvent> uncancelEvents, final CallContext context) {
eventsDao.inTransaction(new Transaction<Void, EntitlementEventSqlDao>() {
@Override
public Void inTransaction(final EntitlementEventSqlDao transactional, final TransactionStatus status) throws Exception {
+ final UUID subscriptionId = subscription.getId();
EntitlementEvent cancelledEvent = null;
final Date now = clock.getUTCNow().toDate();
final List<EntitlementEvent> events = transactional.getFutureActiveEventForSubscription(subscriptionId.toString(), now);
@@ -361,19 +386,23 @@ public class AuditedEntitlementDao implements EntitlementDao {
if (cancelledEvent != null) {
final List<EntityAudit> eventAudits = new ArrayList<EntityAudit>();
-
final String cancelledEventId = cancelledEvent.getId().toString();
transactional.unactiveEvent(cancelledEventId, context);
+
final Long cancelledRecordId = transactional.getRecordId(cancelledEventId);
eventAudits.add(new EntityAudit(TableName.SUBSCRIPTION_EVENTS, cancelledRecordId, ChangeType.UPDATE));
+ EntitlementEvent prev = getLastStoredEventForSubscription(subscriptionId);
for (final EntitlementEvent cur : uncancelEvents) {
transactional.insertEvent(cur, context);
final Long recordId = transactional.getRecordId(cur.getId().toString());
eventAudits.add(new EntityAudit(TableName.SUBSCRIPTION_EVENTS, recordId, ChangeType.INSERT));
recordFutureNotificationFromTransaction(transactional,
- cur.getEffectiveDate(),
+ subscription,
+ prev,
+ cur,
new EntitlementNotificationKey(cur.getId()));
+ prev = cur;
}
transactional.insertAuditFromTransaction(eventAudits, context);
@@ -384,22 +413,27 @@ public class AuditedEntitlementDao implements EntitlementDao {
}
@Override
- public void changePlan(final UUID subscriptionId, final List<EntitlementEvent> changeEvents, final CallContext context) {
+ public void changePlan(final SubscriptionData subscription, final List<EntitlementEvent> changeEvents, final CallContext context) {
eventsDao.inTransaction(new Transaction<Void, EntitlementEventSqlDao>() {
@Override
public Void inTransaction(final EntitlementEventSqlDao transactional, final TransactionStatus status) throws Exception {
+ final UUID subscriptionId = subscription.getId();
cancelNextChangeEventFromTransaction(subscriptionId, transactional, context);
cancelNextPhaseEventFromTransaction(subscriptionId, transactional, context);
final List<EntityAudit> eventAudits = new ArrayList<EntityAudit>();
+ EntitlementEvent prev = getLastStoredEventForSubscription(subscriptionId);
for (final EntitlementEvent cur : changeEvents) {
transactional.insertEvent(cur, context);
final Long recordId = transactional.getRecordId(cur.getId().toString());
eventAudits.add(new EntityAudit(TableName.SUBSCRIPTION_EVENTS, recordId, ChangeType.INSERT));
recordFutureNotificationFromTransaction(transactional,
- cur.getEffectiveDate(),
+ subscription,
+ prev,
+ cur,
new EntitlementNotificationKey(cur.getId()));
+ prev = cur;
}
transactional.insertAuditFromTransaction(eventAudits, context);
@@ -559,21 +593,26 @@ public class AuditedEntitlementDao implements EntitlementDao {
final SubscriptionBundleData bundleData = curBundle.getData();
for (final SubscriptionMigrationData curSubscription : curBundle.getSubscriptions()) {
-
final SubscriptionData subData = curSubscription.getData();
+ EntitlementEvent prevEvent = getLastStoredEventForSubscription(subData.getId());
for (final EntitlementEvent curEvent : curSubscription.getInitialEvents()) {
transactional.insertEvent(curEvent, context);
recordId = transactional.getRecordId(curEvent.getId().toString());
audits.add(new EntityAudit(TableName.SUBSCRIPTION_EVENTS, recordId, ChangeType.INSERT));
recordFutureNotificationFromTransaction(transactional,
- curEvent.getEffectiveDate(),
+ subData,
+ prevEvent,
+ curEvent,
new EntitlementNotificationKey(curEvent.getId()));
+ prevEvent = curEvent;
}
+
transSubDao.insertSubscription(subData, context);
recordId = transSubDao.getRecordId(subData.getId().toString());
audits.add(new EntityAudit(TableName.SUBSCRIPTIONS, recordId, ChangeType.INSERT));
}
+
transBundleDao.insertBundle(bundleData, context);
recordId = transBundleDao.getRecordId(bundleData.getId().toString());
audits.add(new EntityAudit(TableName.BUNDLES, recordId, ChangeType.INSERT));
@@ -597,12 +636,17 @@ public class AuditedEntitlementDao implements EntitlementDao {
for (final EntitlementEvent event : cur.getInitialEvents()) {
transEventDao.updateVersion(event.getId().toString(), event.getActiveVersion(), context);
}
- for (final EntitlementEvent event : cur.getNewEvents()) {
- transEventDao.insertEvent(event, context);
- if (event.getEffectiveDate().isAfter(clock.getUTCNow())) {
+
+ EntitlementEvent prevEvent = getLastStoredEventForSubscription(cur.getId());
+ for (final EntitlementEvent newEvent : cur.getNewEvents()) {
+ transEventDao.insertEvent(newEvent, context);
+ if (newEvent.getEffectiveDate().isAfter(clock.getUTCNow())) {
recordFutureNotificationFromTransaction(transactional,
- event.getEffectiveDate(),
- new EntitlementNotificationKey(event.getId()));
+ cur,
+ prevEvent,
+ newEvent,
+ new EntitlementNotificationKey(newEvent.getId()));
+ prevEvent = newEvent;
}
}
}
@@ -630,15 +674,55 @@ public class AuditedEntitlementDao implements EntitlementDao {
return null;
}
- private void recordFutureNotificationFromTransaction(final Transmogrifier transactionalDao, final DateTime effectiveDate, final NotificationKey notificationKey) {
+ private void recordFutureNotificationFromTransaction(final Transmogrifier transactionalDao, final SubscriptionData subscription,
+ @Nullable final EntitlementEvent prevEvent, final EntitlementEvent nextEvent,
+ final EntitlementNotificationKey notificationKey) {
try {
+ // Add an event to the notification queue
final NotificationQueue subscriptionEventQueue = notificationQueueService.getNotificationQueue(Engine.ENTITLEMENT_SERVICE_NAME,
Engine.NOTIFICATION_QUEUE_NAME);
- subscriptionEventQueue.recordFutureNotificationFromTransaction(transactionalDao, effectiveDate, notificationKey);
+ subscriptionEventQueue.recordFutureNotificationFromTransaction(transactionalDao, nextEvent.getEffectiveDate(), notificationKey);
+
+ // Send an event to the bus
+ final RequestedSubscriptionEvent requestedSubscriptionEvent = createRequestedSubscriptionEvent(subscription, prevEvent, nextEvent, notificationKey);
+ eventBus.postFromTransaction(requestedSubscriptionEvent, transactionalDao);
} catch (NoSuchNotificationQueue e) {
throw new RuntimeException(e);
} catch (IOException e) {
throw new RuntimeException(e);
+ } catch (EventBusException e) {
+ log.warn("Failed to post future entitlement event for subscription " + nextEvent.getSubscriptionId(), e);
+ }
+ }
+
+ private RequestedSubscriptionEvent createRequestedSubscriptionEvent(final SubscriptionData subscription, @Nullable final EntitlementEvent prevEvent,
+ final EntitlementEvent nextEvent, final EntitlementNotificationKey notificationKey) {
+ // We deep copy the subscription object and rehydrate it with this single pair of events in order to build the future transition event
+ // Note that we are still in the transaction so we can't call the database
+ final SubscriptionData subscriptionCopy = new SubscriptionData(new SubscriptionBuilder(subscription));
+ // Fake the events to be active to build the future transition
+ nextEvent.reactivate();
+
+ final ImmutableList<EntitlementEvent> events;
+ if (prevEvent != null) {
+ prevEvent.reactivate();
+ events = ImmutableList.<EntitlementEvent>of(prevEvent, nextEvent);
+ } else {
+ events = ImmutableList.<EntitlementEvent>of(nextEvent);
}
+ subscriptionCopy.rebuildTransitions(events, catalogService.getFullCatalog());
+
+ final EffectiveSubscriptionEvent transitionFromEvent = subscriptionCopy.getTransitionFromEvent(nextEvent, notificationKey.getSeqId());
+ return new DefaultRequestedSubscriptionEvent(transitionFromEvent);
+ }
+
+ private EntitlementEvent getLastStoredEventForSubscription(final UUID subscriptionId) {
+ final List<EntitlementEvent> events = getEventsForSubscription(subscriptionId);
+ EntitlementEvent event = null;
+ if (events != null && events.size() > 0) {
+ event = events.get(events.size() - 1);
+ }
+
+ return event;
}
}
diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/EntitlementDao.java b/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/EntitlementDao.java
index 710391b..b85a47a 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/EntitlementDao.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/EntitlementDao.java
@@ -56,7 +56,7 @@ public interface EntitlementDao {
public void updateChargedThroughDate(final SubscriptionData subscription, final CallContext context);
// Event apis
- public void createNextPhaseEvent(final UUID subscriptionId, final EntitlementEvent nextPhase, final CallContext context);
+ public void createNextPhaseEvent(final SubscriptionData subscription, final EntitlementEvent nextPhase, final CallContext context);
public EntitlementEvent getEventById(final UUID eventId);
@@ -69,13 +69,13 @@ public interface EntitlementDao {
// Subscription creation, cancellation, changePlan apis
public void createSubscription(final SubscriptionData subscription, final List<EntitlementEvent> initialEvents, final CallContext context);
- public void recreateSubscription(final UUID subscriptionId, final List<EntitlementEvent> recreateEvents, final CallContext context);
+ public void recreateSubscription(final SubscriptionData subscription, final List<EntitlementEvent> recreateEvents, final CallContext context);
- public void cancelSubscription(final UUID subscriptionId, final EntitlementEvent cancelEvent, final CallContext context, final int cancelSeq);
+ public void cancelSubscription(final SubscriptionData subscription, final EntitlementEvent cancelEvent, final CallContext context, final int cancelSeq);
- public void uncancelSubscription(final UUID subscriptionId, final List<EntitlementEvent> uncancelEvents, final CallContext context);
+ public void uncancelSubscription(final SubscriptionData subscription, final List<EntitlementEvent> uncancelEvents, final CallContext context);
- public void changePlan(final UUID subscriptionId, final List<EntitlementEvent> changeEvents, final CallContext context);
+ public void changePlan(final SubscriptionData subscription, final List<EntitlementEvent> changeEvents, final CallContext context);
public void migrate(final UUID accountId, final AccountMigrationData data, final CallContext context);
diff --git a/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/RepairEntitlementDao.java b/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/RepairEntitlementDao.java
index 653e1ca..8ffec41 100644
--- a/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/RepairEntitlementDao.java
+++ b/entitlement/src/main/java/com/ning/billing/entitlement/engine/dao/RepairEntitlementDao.java
@@ -86,22 +86,22 @@ public class RepairEntitlementDao implements EntitlementDao, RepairEntitlementLi
}
@Override
- public void createSubscription(final SubscriptionData subscription,
- final List<EntitlementEvent> createEvents, final CallContext context) {
+ public void createSubscription(final SubscriptionData subscription, final List<EntitlementEvent> createEvents, final CallContext context) {
addEvents(subscription.getId(), createEvents);
}
@Override
- public void recreateSubscription(final UUID subscriptionId,
- final List<EntitlementEvent> recreateEvents, final CallContext context) {
- addEvents(subscriptionId, recreateEvents);
+ public void recreateSubscription(final SubscriptionData subscription, final List<EntitlementEvent> recreateEvents, final CallContext context) {
+ addEvents(subscription.getId(), recreateEvents);
}
@Override
- public void cancelSubscription(final UUID subscriptionId,
- final EntitlementEvent cancelEvent, final CallContext context, final int cancelSeq) {
+ public void cancelSubscription(final SubscriptionData subscription, final EntitlementEvent cancelEvent, final CallContext context, final int cancelSeq) {
+ final UUID subscriptionId = subscription.getId();
final long activeVersion = cancelEvent.getActiveVersion();
+
addEvents(subscriptionId, Collections.singletonList(cancelEvent));
+
final SubscriptionRepairEvent target = getRepairSubscriptionEvents(subscriptionId);
boolean foundCancelEvent = false;
for (final EntitlementEvent cur : target.getEvents()) {
@@ -114,9 +114,8 @@ public class RepairEntitlementDao implements EntitlementDao, RepairEntitlementLi
}
@Override
- public void changePlan(final UUID subscriptionId,
- final List<EntitlementEvent> changeEvents, final CallContext context) {
- addEvents(subscriptionId, changeEvents);
+ public void changePlan(final SubscriptionData subscription, final List<EntitlementEvent> changeEvents, final CallContext context) {
+ addEvents(subscription.getId(), changeEvents);
}
@Override
@@ -142,8 +141,7 @@ public class RepairEntitlementDao implements EntitlementDao, RepairEntitlementLi
}
@Override
- public void uncancelSubscription(final UUID subscriptionId,
- final List<EntitlementEvent> uncancelEvents, final CallContext context) {
+ public void uncancelSubscription(final SubscriptionData subscription, final List<EntitlementEvent> uncancelEvents, final CallContext context) {
throw new EntitlementError(NOT_IMPLEMENTED);
}
@@ -163,14 +161,12 @@ public class RepairEntitlementDao implements EntitlementDao, RepairEntitlementLi
}
@Override
- public SubscriptionBundle createSubscriptionBundle(
- final SubscriptionBundleData bundle, final CallContext context) {
+ public SubscriptionBundle createSubscriptionBundle(final SubscriptionBundleData bundle, final CallContext context) {
throw new EntitlementError(NOT_IMPLEMENTED);
}
@Override
- public Subscription getSubscriptionFromId(final SubscriptionFactory factory,
- final UUID subscriptionId) {
+ public Subscription getSubscriptionFromId(final SubscriptionFactory factory, final UUID subscriptionId) {
throw new EntitlementError(NOT_IMPLEMENTED);
}
@@ -180,32 +176,27 @@ public class RepairEntitlementDao implements EntitlementDao, RepairEntitlementLi
}
@Override
- public Subscription getBaseSubscription(final SubscriptionFactory factory,
- final UUID bundleId) {
+ public Subscription getBaseSubscription(final SubscriptionFactory factory, final UUID bundleId) {
throw new EntitlementError(NOT_IMPLEMENTED);
}
@Override
- public List<Subscription> getSubscriptions(final SubscriptionFactory factory,
- final UUID bundleId) {
+ public List<Subscription> getSubscriptions(final SubscriptionFactory factory, final UUID bundleId) {
throw new EntitlementError(NOT_IMPLEMENTED);
}
@Override
- public List<Subscription> getSubscriptionsForKey(
- final SubscriptionFactory factory, final String bundleKey) {
+ public List<Subscription> getSubscriptionsForKey(final SubscriptionFactory factory, final String bundleKey) {
throw new EntitlementError(NOT_IMPLEMENTED);
}
@Override
- public void updateChargedThroughDate(final SubscriptionData subscription,
- final CallContext context) {
+ public void updateChargedThroughDate(final SubscriptionData subscription, final CallContext context) {
throw new EntitlementError(NOT_IMPLEMENTED);
}
@Override
- public void createNextPhaseEvent(final UUID subscriptionId,
- final EntitlementEvent nextPhase, final CallContext context) {
+ public void createNextPhaseEvent(final SubscriptionData subscription, final EntitlementEvent nextPhase, final CallContext context) {
throw new EntitlementError(NOT_IMPLEMENTED);
}
@@ -220,20 +211,17 @@ public class RepairEntitlementDao implements EntitlementDao, RepairEntitlementLi
}
@Override
- public List<EntitlementEvent> getPendingEventsForSubscription(
- final UUID subscriptionId) {
+ public List<EntitlementEvent> getPendingEventsForSubscription(final UUID subscriptionId) {
throw new EntitlementError(NOT_IMPLEMENTED);
}
@Override
- public void migrate(final UUID accountId, final AccountMigrationData data,
- final CallContext context) {
+ public void migrate(final UUID accountId, final AccountMigrationData data, final CallContext context) {
throw new EntitlementError(NOT_IMPLEMENTED);
}
@Override
- public void repair(final UUID accountId, final UUID bundleId, final List<SubscriptionDataRepair> inRepair,
- final CallContext context) {
+ public void repair(final UUID accountId, final UUID bundleId, final List<SubscriptionDataRepair> inRepair, final CallContext context) {
throw new EntitlementError(NOT_IMPLEMENTED);
}
}
diff --git a/entitlement/src/test/java/com/ning/billing/entitlement/engine/dao/MockEntitlementDaoMemory.java b/entitlement/src/test/java/com/ning/billing/entitlement/engine/dao/MockEntitlementDaoMemory.java
index ec5c17e..582a7ba 100644
--- a/entitlement/src/test/java/com/ning/billing/entitlement/engine/dao/MockEntitlementDaoMemory.java
+++ b/entitlement/src/test/java/com/ning/billing/entitlement/engine/dao/MockEntitlementDaoMemory.java
@@ -169,9 +169,7 @@ public class MockEntitlementDaoMemory implements EntitlementDao, MockEntitlement
}
@Override
- public void recreateSubscription(final UUID subscriptionId,
- final List<EntitlementEvent> recreateEvents, final CallContext context) {
-
+ public void recreateSubscription(final SubscriptionData subscription, final List<EntitlementEvent> recreateEvents, final CallContext context) {
synchronized (events) {
events.addAll(recreateEvents);
for (final EntitlementEvent cur : recreateEvents) {
@@ -233,9 +231,9 @@ public class MockEntitlementDaoMemory implements EntitlementDao, MockEntitlement
}
@Override
- public void createNextPhaseEvent(final UUID subscriptionId, final EntitlementEvent nextPhase,
+ public void createNextPhaseEvent(final SubscriptionData subscription, final EntitlementEvent nextPhase,
final CallContext context) {
- cancelNextPhaseEvent(subscriptionId);
+ cancelNextPhaseEvent(subscription.getId());
insertEvent(nextPhase);
}
@@ -271,18 +269,19 @@ public class MockEntitlementDaoMemory implements EntitlementDao, MockEntitlement
}
@Override
- public void cancelSubscription(final UUID subscriptionId, final EntitlementEvent cancelEvent,
+ public void cancelSubscription(final SubscriptionData subscription, final EntitlementEvent cancelEvent,
final CallContext context, final int seqId) {
synchronized (events) {
- cancelNextPhaseEvent(subscriptionId);
+ cancelNextPhaseEvent(subscription.getId());
insertEvent(cancelEvent);
}
}
@Override
- public void changePlan(final UUID subscriptionId, final List<EntitlementEvent> changeEvents,
+ public void changePlan(final SubscriptionData subscription, final List<EntitlementEvent> changeEvents,
final CallContext context) {
synchronized (events) {
+ final UUID subscriptionId = subscription.getId();
cancelNextChangeEvent(subscriptionId);
cancelNextPhaseEvent(subscriptionId);
events.addAll(changeEvents);
@@ -348,7 +347,7 @@ public class MockEntitlementDaoMemory implements EntitlementDao, MockEntitlement
}
@Override
- public void uncancelSubscription(final UUID subscriptionId, final List<EntitlementEvent> uncancelEvents,
+ public void uncancelSubscription(final SubscriptionData subscription, final List<EntitlementEvent> uncancelEvents,
final CallContext context) {
synchronized (events) {
@@ -356,7 +355,7 @@ public class MockEntitlementDaoMemory implements EntitlementDao, MockEntitlement
final Iterator<EntitlementEvent> it = events.descendingIterator();
while (it.hasNext()) {
final EntitlementEvent cur = it.next();
- if (cur.getSubscriptionId() != subscriptionId) {
+ if (cur.getSubscriptionId() != subscription.getId()) {
continue;
}
if (cur.getType() == EventType.API_USER &&
diff --git a/entitlement/src/test/java/com/ning/billing/entitlement/engine/dao/MockEntitlementDaoSql.java b/entitlement/src/test/java/com/ning/billing/entitlement/engine/dao/MockEntitlementDaoSql.java
index 929f03d..6a93d3c 100644
--- a/entitlement/src/test/java/com/ning/billing/entitlement/engine/dao/MockEntitlementDaoSql.java
+++ b/entitlement/src/test/java/com/ning/billing/entitlement/engine/dao/MockEntitlementDaoSql.java
@@ -24,6 +24,7 @@ import org.skife.jdbi.v2.sqlobject.mixins.CloseMe;
import org.skife.jdbi.v2.sqlobject.mixins.Transactional;
import com.google.inject.Inject;
+import com.ning.billing.catalog.api.CatalogService;
import com.ning.billing.entitlement.engine.addon.AddonUtils;
import com.ning.billing.util.bus.Bus;
import com.ning.billing.util.clock.Clock;
@@ -35,8 +36,8 @@ public class MockEntitlementDaoSql extends AuditedEntitlementDao implements Mock
@Inject
public MockEntitlementDaoSql(final IDBI dbi, final Clock clock, final AddonUtils addonUtils, final NotificationQueueService notificationQueueService,
- final Bus eventBus) {
- super(dbi, clock, addonUtils, notificationQueueService, eventBus);
+ final Bus eventBus, final CatalogService catalogService) {
+ super(dbi, clock, addonUtils, notificationQueueService, eventBus, catalogService);
this.resetDao = dbi.onDemand(ResetSqlDao.class);
}