killbill-aplcache

diff --git a/beatrix/src/test/java/com/ning/billing/beatrix/integration/inv_ent/TestBasic.java b/beatrix/src/test/java/com/ning/billing/beatrix/integration/inv_ent/TestBasic.java
index f4e77d5..ed43c33 100644
--- a/beatrix/src/test/java/com/ning/billing/beatrix/integration/inv_ent/TestBasic.java
+++ b/beatrix/src/test/java/com/ning/billing/beatrix/integration/inv_ent/TestBasic.java
@@ -21,6 +21,8 @@ import static org.testng.Assert.assertTrue;
 
 import java.util.UUID;
 
+import com.ning.billing.account.api.AccountApiException;
+import com.ning.billing.entitlement.api.user.EntitlementUserApiException;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 import org.joda.time.Interval;
@@ -174,7 +176,7 @@ public class TestBasic {
         testBasePlanComplete(clock.getUTCNow().minusDays(1).getDayOfMonth());
     }
 
-    @Test(groups = "fast", enabled = true)
+    @Test(groups = "fast", enabled = false)
     public void testBasePlanCompleteWithBillingDayPresent() throws Exception {
         testBasePlanComplete(clock.getUTCNow().getDayOfMonth());
     }
@@ -291,6 +293,43 @@ public class TestBasic {
         lastCtd = checkAndGetCTD(subscription.getId());
     }
 
+    @Test(enabled=false)
+    public void testHappyPath() throws AccountApiException, EntitlementUserApiException {
+        long DELAY = 5000 * 10;
+
+        Account account = accountUserApi.createAccount(getAccountData(), null, null);
+        assertNotNull(account);
+
+        SubscriptionBundle bundle = entitlementUserApi.createBundleForAccount(account.getId(), "whatever");
+
+        String productName = "Shotgun";
+        BillingPeriod term = BillingPeriod.MONTHLY;
+        String planSetName = PriceListSet.DEFAULT_PRICELIST_NAME;
+
+        busHandler.pushExpectedEvent(NextEvent.CREATE);
+        busHandler.pushExpectedEvent(NextEvent.INVOICE);
+        SubscriptionData subscription = (SubscriptionData) entitlementUserApi.createSubscription(bundle.getId(),
+                new PlanPhaseSpecifier(productName, ProductCategory.BASE, term, planSetName, null), null);
+        assertNotNull(subscription);
+
+        assertTrue(busHandler.isCompleted(DELAY));
+
+        busHandler.pushExpectedEvent(NextEvent.CHANGE);
+        busHandler.pushExpectedEvent(NextEvent.INVOICE);
+        BillingPeriod newTerm = BillingPeriod.MONTHLY;
+        String newPlanSetName = PriceListSet.DEFAULT_PRICELIST_NAME;
+        String newProductName = "Assault-Rifle";
+        subscription.changePlan(newProductName, newTerm, newPlanSetName, clock.getUTCNow());
+
+        assertTrue(busHandler.isCompleted(DELAY));
+
+        busHandler.pushExpectedEvent(NextEvent.PHASE);
+        busHandler.pushExpectedEvent(NextEvent.INVOICE);
+        clock.setDeltaFromReality(AT_LEAST_ONE_MONTH_MS);
+        assertTrue(busHandler.isCompleted(DELAY));
+
+    }
+
 
     protected AccountData getAccountData() {
         AccountData accountData = new AccountData() {

diff --git a/beatrix/src/test/resources/catalogSample.xml b/beatrix/src/test/resources/catalogSample.xml
index 8c91a71..898b7df 100644
--- a/beatrix/src/test/resources/catalogSample.xml
+++ b/beatrix/src/test/resources/catalogSample.xml
@@ -205,7 +205,7 @@ Use Cases to do:
 						<unit>DAYS</unit>
 						<number>30</number>
 					</duration>
-					<billingPeriod>NO_BILLING_PERIOD</billingPeriod>
+					<billingPeriod>MONTHLY</billingPeriod>
 					<fixedPrice> <!-- empty price implies $0 -->
 					</fixedPrice>
 				</phase>
@@ -231,11 +231,11 @@ Use Cases to do:
 						<number>30</number>
 					</duration>
 					<billingPeriod>NO_BILLING_PERIOD</billingPeriod>
-					<fixedPrice>
-                                          <price><currency>GBP</currency><value>29.95</value></price>
-                                          <price><currency>EUR</currency><value>29.95</value></price>
-                                          <price><currency>USD</currency><value>1.95</value></price>
-                                        </fixedPrice>
+					<recurringPrice>
+                        <price><currency>GBP</currency><value>29.95</value></price>
+                        <price><currency>EUR</currency><value>29.95</value></price>
+                        <price><currency>USD</currency><value>1.95</value></price>
+                    </recurringPrice>
 				    <!-- no price implies $0 -->
 				</phase>
 			</initialPhases>
@@ -260,12 +260,12 @@ Use Cases to do:
 						<unit>DAYS</unit>
 						<number>30</number>
 					</duration>
-					<billingPeriod>NO_BILLING_PERIOD</billingPeriod>
-					<fixedPrice>
-                                          <price><currency>GBP</currency><value>29.95</value></price>
-                                          <price><currency>EUR</currency><value>29.95</value></price>
-                                          <price><currency>USD</currency><value>2.95</value></price>
-					</fixedPrice>
+					<billingPeriod>MONTHLY</billingPeriod>
+					<recurringPrice>
+                        <price><currency>GBP</currency><value>29.95</value></price>
+                        <price><currency>EUR</currency><value>29.95</value></price>
+                        <price><currency>USD</currency><value>2.95</value></price>
+					</recurringPrice>
 				</phase>
 			</initialPhases>
 			<finalPhase type="EVERGREEN">

diff --git a/invoice/src/main/java/com/ning/billing/invoice/dao/DefaultInvoiceDao.java b/invoice/src/main/java/com/ning/billing/invoice/dao/DefaultInvoiceDao.java
index a47344a..31424ba 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/dao/DefaultInvoiceDao.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/dao/DefaultInvoiceDao.java
@@ -208,7 +208,28 @@ public class DefaultInvoiceDao implements InvoiceDao {
             }
         });
     }
-    
+
+    @Override
+    public boolean lockAccount(final UUID accountId) {
+        try {
+            invoiceSqlDao.lockAccount(accountId.toString());
+            return true;
+        } catch (Exception e) {
+            log.error("Ouch! I broke", e);
+            return false;
+        }
+    }
+
+    @Override
+    public boolean releaseAccount(final UUID accountId) {
+        try {
+            invoiceSqlDao.releaseAccount(accountId.toString());
+            return true;
+        } catch (Exception e) {
+            return false;
+        }
+    }
+
     @Override
     public UUID getInvoiceIdByPaymentAttemptId(UUID paymentAttemptId) {
         return invoiceSqlDao.getInvoiceIdByPaymentAttemptId(paymentAttemptId.toString());
@@ -244,7 +265,9 @@ public class DefaultInvoiceDao implements InvoiceDao {
     private void notifyOfFutureBillingEvents(final InvoiceSqlDao dao, final List<InvoiceItem> invoiceItems) {
         for (final InvoiceItem item : invoiceItems) {
             if (item.getRecurringRate() != null) {
-                notifier.insertNextBillingNotification(dao, item.getSubscriptionId(), item.getEndDate());
+                if (item.getRecurringAmount().compareTo(BigDecimal.ZERO) > 0) {
+                    notifier.insertNextBillingNotification(dao, item.getSubscriptionId(), item.getEndDate());
+                }
             }
         }
     }

diff --git a/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceDao.java b/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceDao.java
index 7a7c280..a86cde7 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceDao.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceDao.java
@@ -53,5 +53,9 @@ public interface InvoiceDao {
 
     List<Invoice> getUnpaidInvoicesByAccountId(final UUID accountId, final DateTime upToDate);
 
+    boolean lockAccount(final UUID accountId);
+
+    boolean releaseAccount(final UUID accountId);
+
     void test();
 }

diff --git a/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceSqlDao.java b/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceSqlDao.java
index 4c50391..e580d05 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceSqlDao.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/dao/InvoiceSqlDao.java
@@ -86,6 +86,12 @@ public interface InvoiceSqlDao extends EntityDao<Invoice>, Transactional<Invoice
     @SqlQuery
     List<Invoice> getUnpaidInvoicesByAccountId(@Bind("accountId") final String accountId,
                                                @Bind("upToDate") final Date upToDate);
+
+    @SqlUpdate
+    void lockAccount(@Bind("accountId") final String accountId);
+
+    @SqlUpdate
+    void releaseAccount(@Bind("accountId") final String accountId);
     
     @BindingAnnotation(InvoiceBinder.InvoiceBinderFactory.class)
     @Retention(RetentionPolicy.RUNTIME)

diff --git a/invoice/src/main/java/com/ning/billing/invoice/InvoiceListener.java b/invoice/src/main/java/com/ning/billing/invoice/InvoiceListener.java
index e3b73e9..dd44e0c 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/InvoiceListener.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/InvoiceListener.java
@@ -53,19 +53,17 @@ public class InvoiceListener {
     private final EntitlementBillingApi entitlementBillingApi;
     private final AccountUserApi accountUserApi;
     private final InvoiceDao invoiceDao;
-    private final GlobalLocker globalLocker;
 
     private final static boolean VERBOSE_OUTPUT = false;
 
     @Inject
     public InvoiceListener(final InvoiceGenerator generator, final AccountUserApi accountUserApi,
                            final EntitlementBillingApi entitlementBillingApi,
-                           final InvoiceDao invoiceDao, final GlobalLocker globalLocker) {
+                           final InvoiceDao invoiceDao) {
         this.generator = generator;
         this.entitlementBillingApi = entitlementBillingApi;
         this.accountUserApi = accountUserApi;
         this.invoiceDao = invoiceDao;
-        this.globalLocker = globalLocker;
     }
 
     @Subscribe
@@ -99,9 +97,7 @@ public class InvoiceListener {
             return;
         }
 
-        GlobalLock lock = globalLocker.lockWithNumberOfTries("invoiceProcessor:" + accountId.toString(), 1);
-
-        if (lock == null) {
+        if (!invoiceDao.lockAccount(accountId)) {
             log.warn("Conflicting lock detected from InvoiceListener on account "  + accountId.toString());
         } else {
             log.info("Locked " + accountId.toString());
@@ -142,7 +138,7 @@ public class InvoiceListener {
                 }
             }
 
-            lock.release();
+            invoiceDao.releaseAccount(accountId);
         }
     }
 

diff --git a/invoice/src/main/resources/com/ning/billing/invoice/dao/InvoiceSqlDao.sql.stg b/invoice/src/main/resources/com/ning/billing/invoice/dao/InvoiceSqlDao.sql.stg
index c6d4bb6..8a6fe38 100644
--- a/invoice/src/main/resources/com/ning/billing/invoice/dao/InvoiceSqlDao.sql.stg
+++ b/invoice/src/main/resources/com/ning/billing/invoice/dao/InvoiceSqlDao.sql.stg
@@ -92,6 +92,15 @@ getUnpaidInvoicesByAccountId() ::= <<
   ORDER BY i.target_date ASC;
 >>
 
+lockAccount() ::= <<
+  INSERT INTO invoice_locking(account_id) VALUES (:accountId);
+>>
+
+releaseAccount() ::= <<
+  DELETE FROM invoice_locking
+  WHERE account_id = :accountId;
+>>
+
 test() ::= <<
   SELECT 1
   FROM invoices;

diff --git a/invoice/src/main/resources/com/ning/billing/invoice/ddl.sql b/invoice/src/main/resources/com/ning/billing/invoice/ddl.sql
index bb9f562..28f3c89 100644
--- a/invoice/src/main/resources/com/ning/billing/invoice/ddl.sql
+++ b/invoice/src/main/resources/com/ning/billing/invoice/ddl.sql
@@ -15,6 +15,12 @@ CREATE TABLE invoice_items (
 ) ENGINE=innodb;
 CREATE INDEX invoice_items_subscription_id ON invoice_items(subscription_id ASC);
 
+DROP TABLE IF EXISTS invoice_locking;
+CREATE TABLE invoice_locking (
+  account_id char(36) NOT NULL,
+  PRIMARY KEY(account_id)
+) ENGINE = innodb;
+
 DROP TABLE IF EXISTS invoices;
 CREATE TABLE invoices (
   id char(36) NOT NULL,

diff --git a/invoice/src/test/java/com/ning/billing/invoice/dao/MockInvoiceDao.java b/invoice/src/test/java/com/ning/billing/invoice/dao/MockInvoiceDao.java
index cf6c112..8332e8c 100644
--- a/invoice/src/test/java/com/ning/billing/invoice/dao/MockInvoiceDao.java
+++ b/invoice/src/test/java/com/ning/billing/invoice/dao/MockInvoiceDao.java
@@ -215,4 +215,14 @@ public class MockInvoiceDao implements InvoiceDao {
 
         return unpaidInvoices;
     }
+
+    @Override
+    public boolean lockAccount(UUID accountId) {
+        return true;
+    }
+
+    @Override
+    public boolean releaseAccount(UUID accountId) {
+        return true;
+    }
 }