Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

listener: refresh payment rows when CBA is distributed #106

Merged
merged 1 commit into from
Nov 5, 2019
Merged

listener: refresh payment rows when CBA is distributed #106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 2 additions & 4 deletions src/main/java/org/killbill/billing/plugin/analytics/AnalyticsJobHierarchy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,10 @@

package org.killbill.billing.plugin.analytics;

import org.killbill.billing.notification.plugin.api.ExtBusEventType;

public abstract class AnalyticsJobHierarchy {

public static Group fromEventType(final ExtBusEventType eventType) {
switch (eventType) {
public static Group fromEventType(final AnalyticsJob job) {
switch (job.getEventType()) {
// Account information is denormalized across all tables
case ACCOUNT_CREATION:
case ACCOUNT_CHANGE:
Expand Down
39 changes: 31 additions & 8 deletions src/main/java/org/killbill/billing/plugin/analytics/AnalyticsListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.killbill.billing.plugin.analytics;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.Iterator;
import java.util.UUID;
import java.util.concurrent.Executor;
Expand All @@ -27,6 +28,8 @@

import org.joda.time.DateTime;
import org.killbill.billing.ObjectType;
import org.killbill.billing.invoice.api.Invoice;
import org.killbill.billing.invoice.api.InvoiceApiException;
import org.killbill.billing.notification.plugin.api.ExtBusEvent;
import org.killbill.billing.osgi.libs.killbill.OSGIConfigPropertiesService;
import org.killbill.billing.osgi.libs.killbill.OSGIKillbillAPI;
Expand All @@ -43,9 +46,11 @@
import org.killbill.billing.plugin.analytics.dao.BusinessSubscriptionTransitionDao;
import org.killbill.billing.plugin.analytics.dao.CurrencyConversionDao;
import org.killbill.billing.plugin.analytics.dao.factory.BusinessContextFactory;
import org.killbill.billing.plugin.api.PluginTenantContext;
import org.killbill.billing.util.api.RecordIdApi;
import org.killbill.billing.util.callcontext.CallContext;
import org.killbill.billing.util.callcontext.CallOrigin;
import org.killbill.billing.util.callcontext.TenantContext;
import org.killbill.billing.util.callcontext.UserType;
import org.killbill.clock.Clock;
import org.killbill.notificationq.DefaultNotificationQueueService;
Expand All @@ -66,6 +71,7 @@
import com.google.common.collect.Iterables;
import com.google.common.collect.Iterators;

import static org.killbill.billing.notification.plugin.api.ExtBusEventType.PAYMENT_SUCCESS;
import static org.killbill.billing.plugin.analytics.AnalyticsActivator.ANALYTICS_QUEUE_SERVICE;

public class AnalyticsListener implements OSGIKillbillEventDispatcher.OSGIKillbillEventHandler {
Expand Down Expand Up @@ -186,13 +192,30 @@ public void handleKillbillEvent(final ExtBusEvent killbillEvent) {
return;
}

AnalyticsJob job = new AnalyticsJob(killbillEvent);
if (AnalyticsJobHierarchy.fromEventType(job) == Group.INVOICES) {
try {
final TenantContext tenantContext = new PluginTenantContext(killbillEvent.getAccountId(), killbillEvent.getTenantId());
final Invoice invoice = osgiKillbillAPI.getInvoiceUserApi().getInvoice(killbillEvent.getObjectId(), tenantContext);
if (BigDecimal.ZERO.compareTo(invoice.getCreditedAmount()) != 0 && invoice.getNumberOfPayments() > 0) {
// The invoice has payments and CBA was updated: payment rows must be updated
// See https://github.com/killbill/killbill-analytics-plugin/issues/105
job = new AnalyticsJob(PAYMENT_SUCCESS,
killbillEvent.getObjectType(),
killbillEvent.getObjectId(),
killbillEvent.getAccountId(),
killbillEvent.getTenantId());
}
} catch (final InvoiceApiException e) {
logger.warn("Unable to retrieve InvoiceUserApi for event {}, payment data might be stale", killbillEvent);
}
}

// Events we don't care about
if (shouldIgnoreEvent(killbillEvent)) {
if (shouldIgnoreEvent(job)) {
return;
}

final AnalyticsJob job = new AnalyticsJob(killbillEvent);

Long accountRecordId = null;
Long tenantRecordId = null;
final RecordIdApi recordIdApi = osgiKillbillAPI.getRecordIdApi();
Expand Down Expand Up @@ -273,10 +296,10 @@ public boolean apply(final NotificationEventWithMetadata<AnalyticsJob> notificat
// Does the specified existing notification overlap with this job?
private boolean jobsOverlap(final AnalyticsJob job, final NotificationEventWithMetadata<AnalyticsJob> notificationEventForExistingJob) {
final AnalyticsJob existingJob = notificationEventForExistingJob.getEvent();
final AnalyticsJobHierarchy.Group existingHierarchyGroup = AnalyticsJobHierarchy.fromEventType(existingJob.getEventType());
final AnalyticsJobHierarchy.Group existingHierarchyGroup = AnalyticsJobHierarchy.fromEventType(existingJob);

return existingJob.getAccountId().equals(job.getAccountId()) &&
(existingHierarchyGroup.equals(AnalyticsJobHierarchy.fromEventType(job.getEventType())) ||
(existingHierarchyGroup.equals(AnalyticsJobHierarchy.fromEventType(job)) ||
AnalyticsJobHierarchy.Group.ALL.equals(existingHierarchyGroup));
}

Expand All @@ -288,7 +311,7 @@ private void handleAnalyticsJob(final AnalyticsJob job) throws AnalyticsRefreshE
final CallContext callContext = new AnalyticsCallContext(job, clock);
final BusinessContextFactory businessContextFactory = new BusinessContextFactory(job.getAccountId(), callContext, currencyConversionDao, osgiKillbillAPI, osgiConfigPropertiesService, clock, analyticsConfigurationHandler);

final Group group = AnalyticsJobHierarchy.fromEventType(job.getEventType());
final Group group = AnalyticsJobHierarchy.fromEventType(job);
logger.info("Starting {} Analytics refresh for account {}", group, businessContextFactory.getAccountId());
switch (group) {
case ALL:
Expand Down Expand Up @@ -326,8 +349,8 @@ protected boolean isAccountBlacklisted(@Nullable final UUID accountId) {
}

@VisibleForTesting
protected boolean shouldIgnoreEvent(final ExtBusEvent event) {
return Iterables.find(ignoredGroups, Predicates.<Group>equalTo(AnalyticsJobHierarchy.fromEventType(event.getEventType())), null) != null;
protected boolean shouldIgnoreEvent(final AnalyticsJob job) {
return Iterables.find(ignoredGroups, Predicates.<Group>equalTo(AnalyticsJobHierarchy.fromEventType(job)), null) != null;
}

@VisibleForTesting
Expand Down
4 changes: 2 additions & 2 deletions src/test/java/org/killbill/billing/plugin/analytics/TestAnalyticsListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@ public Object answer(InvocationOnMock invocation) throws Throwable {

final AnalyticsListener analyticsListener = new AnalyticsListener(killbillAPI, killbillDataSource, osgiConfigPropertiesService, null, clock, analyticsConfigurationHandler, notificationQueueService);

Assert.assertTrue(analyticsListener.shouldIgnoreEvent(cfEvent));
Assert.assertFalse(analyticsListener.shouldIgnoreEvent(accountEvent));
Assert.assertTrue(analyticsListener.shouldIgnoreEvent(new AnalyticsJob(cfEvent)));
Assert.assertFalse(analyticsListener.shouldIgnoreEvent(new AnalyticsJob(accountEvent)));
}
}