killbill-aplcache

diff --git a/beatrix/src/test/java/com/ning/billing/beatrix/integration/overdue/TestOverdueIntegration.java b/beatrix/src/test/java/com/ning/billing/beatrix/integration/overdue/TestOverdueIntegration.java
index 0242954..8480faf 100644
--- a/beatrix/src/test/java/com/ning/billing/beatrix/integration/overdue/TestOverdueIntegration.java
+++ b/beatrix/src/test/java/com/ning/billing/beatrix/integration/overdue/TestOverdueIntegration.java
@@ -23,6 +23,7 @@ import static org.testng.Assert.assertTrue;
 
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
+import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -218,19 +219,11 @@ public class TestOverdueIntegration extends TestIntegrationBase {
         paymentPlugin.makeAllInvoicesFail(false);
         Collection<Invoice> invoices = invoiceApi.getUnpaidInvoicesByAccountId(account.getId(), clock.getUTCNow());
         List<String> invoiceIds = new ArrayList<String>();
-        List<PaymentAttempt> paymentAttempts = new ArrayList<PaymentAttempt>();
         for(Invoice invoice : invoices) {
-//            paymentAttempts.addAll(paymentApi.getPaymentAttemptsForInvoiceId(invoice.getId().toString()));
-//            for(PaymentAttempt pa : paymentAttempts) {
-//                busHandler.pushExpectedEvent(NextEvent.PAYMENT);
-//                try {
-//                paymentApi.createPaymentForPaymentAttempt(pa.getPaymentAttemptId(), new DefaultCallContext("test", null, null, clock));
-//                } catch(PaymentApiException e) {
-//                    log.info("",e);
-//                }
-//            }
-            invoiceIds.add(invoice.getId().toString());
-            
+            invoiceIds.add(invoice.getId().toString()); 
+            if(invoice.getBalance().compareTo(BigDecimal.ZERO) > 0) {
+                busHandler.pushExpectedEvent(NextEvent.PAYMENT);
+            }
         }
         paymentApi.createPayment(account, invoiceIds, new DefaultCallContext("test", null, null, clock));
        

diff --git a/payment/src/main/java/com/ning/billing/payment/api/DefaultPaymentApi.java b/payment/src/main/java/com/ning/billing/payment/api/DefaultPaymentApi.java
index 2ac89d2..a670ab3 100644
--- a/payment/src/main/java/com/ning/billing/payment/api/DefaultPaymentApi.java
+++ b/payment/src/main/java/com/ning/billing/payment/api/DefaultPaymentApi.java
@@ -41,6 +41,9 @@ import com.ning.billing.payment.RetryService;
 import com.ning.billing.payment.dao.PaymentDao;
 import com.ning.billing.payment.provider.PaymentProviderPlugin;
 import com.ning.billing.payment.provider.PaymentProviderPluginRegistry;
+import com.ning.billing.util.bus.Bus;
+import com.ning.billing.util.bus.BusEvent;
+import com.ning.billing.util.bus.Bus.EventBusException;
 import com.ning.billing.util.callcontext.CallContext;
 
 public class DefaultPaymentApi implements PaymentApi {
@@ -50,6 +53,7 @@ public class DefaultPaymentApi implements PaymentApi {
     private final RetryService retryService;
     private final PaymentDao paymentDao;
     private final PaymentConfig config;
+    private final Bus eventBus;
 
 
     private static final Logger log = LoggerFactory.getLogger(DefaultPaymentApi.class);
@@ -60,13 +64,16 @@ public class DefaultPaymentApi implements PaymentApi {
             InvoicePaymentApi invoicePaymentApi,
             RetryService retryService,
             PaymentDao paymentDao,
-            PaymentConfig config) {
+            PaymentConfig config,
+            Bus eventBus) {
         this.pluginRegistry = pluginRegistry;
         this.accountUserApi = accountUserApi;
         this.invoicePaymentApi = invoicePaymentApi;
         this.retryService = retryService;
         this.paymentDao = paymentDao;
         this.config = config;
+        this.eventBus = eventBus;
+
     }
 
     @Override
@@ -274,6 +281,7 @@ public class DefaultPaymentApi implements PaymentApi {
             log.info("Could not process a payment for " + paymentAttempt + " error was " + error);
 
             scheduleRetry(paymentAttempt, error);
+            postPaymentEvent(paymentOrError.getLeft(), account.getId());
         }
         else {
             paymentInfo = paymentOrError.getRight();
@@ -301,7 +309,8 @@ public class DefaultPaymentApi implements PaymentApi {
             if (paymentInfo.getPaymentId() != null) {
                 paymentDao.updatePaymentAttemptWithPaymentId(paymentAttempt.getPaymentAttemptId(), paymentInfo.getPaymentId(), context);
             }
-        }
+            postPaymentEvent(paymentInfo, account.getId());
+       }
 
         invoicePaymentApi.notifyOfPaymentAttempt(
                 invoice.getId(),
@@ -314,6 +323,17 @@ public class DefaultPaymentApi implements PaymentApi {
 
         return paymentOrError;
     }
+    
+    private void postPaymentEvent(BusEvent ev, UUID accountId) {
+        if (ev == null) {
+            return;
+        }
+        try {
+            eventBus.post(ev);
+        } catch (EventBusException e) {
+            log.error("Failed to post Payment event event for account {} ", accountId, e);
+        }
+    }
 
     private void scheduleRetry(PaymentAttempt paymentAttempt, String error) {
         final List<Integer> retryDays = config.getPaymentRetryDays();

diff --git a/payment/src/main/java/com/ning/billing/payment/RequestProcessor.java b/payment/src/main/java/com/ning/billing/payment/RequestProcessor.java
index c90d859..508fe21 100644
--- a/payment/src/main/java/com/ning/billing/payment/RequestProcessor.java
+++ b/payment/src/main/java/com/ning/billing/payment/RequestProcessor.java
@@ -48,7 +48,6 @@ public class RequestProcessor {
     public static final String PAYMENT_PROVIDER_KEY = "paymentProvider";
     private final AccountUserApi accountUserApi;
     private final PaymentApi paymentApi;
-    private final Bus eventBus;
     private final Clock clock;
 
     private static final Logger log = LoggerFactory.getLogger(RequestProcessor.class);
@@ -57,25 +56,13 @@ public class RequestProcessor {
     public RequestProcessor(Clock clock,
             AccountUserApi accountUserApi,
             PaymentApi paymentApi,
-            PaymentProviderPluginRegistry pluginRegistry,
-            Bus eventBus) {
+            PaymentProviderPluginRegistry pluginRegistry) {
         this.clock = clock;
         this.accountUserApi = accountUserApi;
         this.paymentApi = paymentApi;
-        this.eventBus = eventBus;
-    }
-
-    private void postPaymentEvent(BusEvent ev, UUID accountId) {
-        if (ev == null) {
-            return;
-        }
-        try {
-            eventBus.post(ev);
-        } catch (EventBusException e) {
-            log.error("Failed to post Payment event event for account {} ", accountId, e);
-        }
     }
 
+ 
     @Subscribe
     public void receiveInvoice(InvoiceCreationEvent event) {
 
@@ -88,7 +75,6 @@ public class RequestProcessor {
                 CallContext context = new DefaultCallContext("PaymentRequestProcessor", CallOrigin.INTERNAL, UserType.SYSTEM, clock);
                 List<PaymentInfoEvent> results = paymentApi.createPayment(account, Arrays.asList(event.getInvoiceId().toString()), context);
                 PaymentInfoEvent infoEvent = (!results.isEmpty()) ?  results.get(0) : null;
-                postPaymentEvent(infoEvent, account.getId());
                 return;
             } else {
                 errorEvent = new DefaultPaymentErrorEvent(null, "Failed to retrieve account", event.getAccountId(), event.getInvoiceId(), event.getUserToken());
@@ -100,6 +86,5 @@ public class RequestProcessor {
             log.info("Failed to process invoice payment for account: " + event.getAccountId()); // Removed stack trace log here and changed to info - this an expected flow
             errorEvent = new DefaultPaymentErrorEvent(null, e.getMessage(), event.getAccountId(), event.getInvoiceId(), event.getUserToken());                        
         }
-        postPaymentEvent(errorEvent, event.getAccountId());
     }
 }