Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
subscription: send effective events from transactions
Relates to #451. Signed-off-by: Pierre-Alexandre Meyer <pierre@mouraf.org>
- Loading branch information
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -45,7 +45,6 @@ | |
| import org.killbill.billing.catalog.api.ProductCategory; | ||
| import org.killbill.billing.entitlement.api.SubscriptionApiException; | ||
| import org.killbill.billing.entity.EntityPersistenceException; | ||
| import org.killbill.billing.subscription.api.SubscriptionBase; | ||
| import org.killbill.billing.subscription.api.SubscriptionBaseTransitionType; | ||
| import org.killbill.billing.subscription.api.migration.AccountMigrationData; | ||
|
|
@@ -86,6 +85,7 @@ | |
| import org.killbill.billing.util.entity.dao.EntitySqlDaoTransactionWrapper; | ||
| import org.killbill.billing.util.entity.dao.EntitySqlDaoTransactionalJdbiWrapper; | ||
| import org.killbill.billing.util.entity.dao.EntitySqlDaoWrapperFactory; | ||
| import org.killbill.bus.api.BusEvent; | ||
| import org.killbill.bus.api.PersistentBus; | ||
| import org.killbill.bus.api.PersistentBus.EventBusException; | ||
| import org.killbill.clock.Clock; | ||
|
|
@@ -405,20 +405,21 @@ public Void inTransaction(final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFa | |
| } | ||
|
|
||
| @Override | ||
| public void createNextPhaseEvent(final DefaultSubscriptionBase subscription, final SubscriptionBaseEvent readyPhaseEvent, final SubscriptionBaseEvent nextPhaseEvent, final InternalCallContext context) { | ||
| transactionalSqlDao.execute(new EntitySqlDaoTransactionWrapper<Void>() { | ||
| @Override | ||
| public Void inTransaction(final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFactory) throws Exception { | ||
| final SubscriptionEventSqlDao transactional = entitySqlDaoWrapperFactory.become(SubscriptionEventSqlDao.class); | ||
| final UUID subscriptionId = subscription.getId(); | ||
| cancelNextPhaseEventFromTransaction(subscriptionId, entitySqlDaoWrapperFactory, context); | ||
| transactional.create(new SubscriptionEventModelDao(nextPhaseEvent), context); | ||
| recordFutureNotificationFromTransaction(entitySqlDaoWrapperFactory, | ||
| nextPhaseEvent.getEffectiveDate(), | ||
| new SubscriptionNotificationKey(nextPhaseEvent.getId()), context); | ||
|
|
||
| // Notify the Bus | ||
| notifyBusOfRequestedChange(entitySqlDaoWrapperFactory, subscription, nextPhaseEvent, SubscriptionBaseTransitionType.PHASE, context); | ||
| notifyBusOfEffectiveImmediateChange(entitySqlDaoWrapperFactory, subscription, readyPhaseEvent, 0, context); | ||
|
|
||
| return null; | ||
| } | ||
|
|
@@ -560,6 +561,19 @@ public Void inTransaction(final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFa | |
| }); | ||
| } | ||
|
|
||
| @Override | ||
| public void cancelSubscriptionsOnBasePlanEvent(final DefaultSubscriptionBase subscription, final SubscriptionBaseEvent event, final List<DefaultSubscriptionBase> subscriptions, final List<SubscriptionBaseEvent> cancelEvents, final InternalCallContext context) { | ||
| transactionalSqlDao.execute(new EntitySqlDaoTransactionWrapper<Void>() { | ||
| @Override | ||
| public Void inTransaction(final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFactory) throws Exception { | ||
| cancelSubscriptionsFromTransaction(entitySqlDaoWrapperFactory, subscriptions, cancelEvents, context); | ||
| // Make sure to always send the event, even if there were no subscriptions to cancel | ||
| notifyBusOfEffectiveImmediateChange(entitySqlDaoWrapperFactory, subscription, event, subscriptions.size(), context); | ||
| return null; | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| @Override | ||
| public void cancelSubscriptions(final List<DefaultSubscriptionBase> subscriptions, final List<SubscriptionBaseEvent> cancelEvents, final InternalCallContext context) { | ||
| transactionalSqlDao.execute(new EntitySqlDaoTransactionWrapper<Void>() { | ||
|
|
@@ -1046,12 +1060,12 @@ private SubscriptionBase getBaseSubscription(final UUID bundleId, final boolean | |
| } | ||
|
|
||
| // | ||
| // Either records a notification or sends a bus event if operation is immediate | ||
| // | ||
| private void recordBusOrFutureNotificationFromTransaction(final DefaultSubscriptionBase subscription, final SubscriptionBaseEvent event, final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFactory, final boolean busEvent, | ||
| final int seqId, final InternalCallContext context) { | ||
| if (busEvent) { | ||
| rebuildSubscriptionAndNotifyBusOfEffectiveImmediateChange(entitySqlDaoWrapperFactory, subscription, event, seqId, context); | ||
| } else { | ||
| recordFutureNotificationFromTransaction(entitySqlDaoWrapperFactory, | ||
| event.getEffectiveDate(), | ||
|
|
@@ -1060,25 +1074,29 @@ private void recordBusOrFutureNotificationFromTransaction(final DefaultSubscript | |
| } | ||
| } | ||
|
|
||
| // Sends bus notification for event on effective date -- only used for operation that happen immediately | ||
| private void rebuildSubscriptionAndNotifyBusOfEffectiveImmediateChange(final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFactory, final DefaultSubscriptionBase subscription, | ||
| final SubscriptionBaseEvent immediateEvent, final int seqId, final InternalCallContext context) { | ||
This comment has been minimized.
Sorry, something went wrong. 
sbrossie
Member
|
||
| try { | ||
| final DefaultSubscriptionBase upToDateSubscription = createSubscriptionWithNewEvent(subscription, immediateEvent, context); | ||
| notifyBusOfEffectiveImmediateChange(entitySqlDaoWrapperFactory, upToDateSubscription, immediateEvent, seqId, context); | ||
| } catch (final CatalogApiException e) { | ||
| log.warn("Failed to post effective event for subscription " + subscription.getId(), e); | ||
| } | ||
| } | ||
|
|
||
| private void notifyBusOfEffectiveImmediateChange(final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFactory, final DefaultSubscriptionBase subscription, | ||
| final SubscriptionBaseEvent immediateEvent, final int seqId, final InternalCallContext context) { | ||
| try { | ||
| final SubscriptionBaseTransitionData transition = subscription.getTransitionFromEvent(immediateEvent, seqId); | ||
| final BusEvent busEvent = new DefaultEffectiveSubscriptionEvent(transition, | ||
| subscription.getAlignStartDate(), | ||
| context.getUserToken(), | ||
| context.getAccountRecordId(), | ||
| context.getTenantRecordId()); | ||
|
|
||
| eventBus.postFromTransaction(busEvent, entitySqlDaoWrapperFactory.getHandle().getConnection()); | ||
| } catch (final EventBusException e) { | ||
| log.warn("Failed to post effective event for subscription " + subscription.getId(), e); | ||
| } | ||
| } | ||
|
|
@@ -1087,7 +1105,7 @@ private void notifyBusOfRequestedChange(final EntitySqlDaoWrapperFactory entityS | |
| final SubscriptionBaseEvent nextEvent, final SubscriptionBaseTransitionType transitionType, final InternalCallContext context) { | ||
| try { | ||
| eventBus.postFromTransaction(new DefaultRequestedSubscriptionEvent(subscription, nextEvent, transitionType, context.getAccountRecordId(), context.getTenantRecordId(), context.getUserToken()), entitySqlDaoWrapperFactory.getHandle().getConnection()); | ||
| } catch (final EventBusException e) { | ||
| log.warn("Failed to post requested change event for subscription " + subscription.getId(), e); | ||
| } | ||
| } | ||
|
|
||
Oops, something went wrong.
If the method
onPhaseEventandonBasePlanEvent, why do we then send the event at this level?