diff --git a/invoice/src/main/java/com/ning/billing/invoice/notification/DefaultNextBillingDateNotifier.java b/invoice/src/main/java/com/ning/billing/invoice/notification/DefaultNextBillingDateNotifier.java
new file mode 100644
index 0000000..ebf0b5b
--- /dev/null
+++ b/invoice/src/main/java/com/ning/billing/invoice/notification/DefaultNextBillingDateNotifier.java
@@ -0,0 +1,135 @@
+/*
+ * Copyright 2010-2011 Ning, Inc.
+ *
+ * Ning licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.ning.billing.invoice.notification;
+
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.skife.jdbi.v2.sqlobject.mixins.Transmogrifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.ning.billing.config.InvoiceConfig;
+import com.ning.billing.lifecycle.KillbillService;
+import com.ning.billing.lifecycle.LifecycleHandlerType;
+import com.ning.billing.lifecycle.LifecycleHandlerType.LifecycleLevel;
+import com.ning.billing.util.eventbus.Bus;
+import com.ning.billing.util.eventbus.Bus.EventBusException;
+import com.ning.billing.util.notificationq.NotificationConfig;
+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.NotficationQueueAlreadyExists;
+import com.ning.billing.util.notificationq.NotificationQueueService.NotificationQueueHandler;
+
+public class DefaultNextBillingDateNotifier implements KillbillService, NextBillingDateNotifier {
+ private final static Logger log = LoggerFactory.getLogger(DefaultNextBillingDateNotifier.class);
+
+ private static final String NEXT_BILLING_DATE_NOTIFIER_SERVICE_NAME = "next-billing-date-notifier";
+ private static final String NEXT_BILLING_DATE_NOTIFIER_QUEUE = "next-billing-date-queue";
+
+ private final Bus eventBus;
+ private final NotificationQueueService notificationQueueService;
+ private NotificationQueue nextBillingQueue;
+ private InvoiceConfig config;
+
+ @Inject
+ public DefaultNextBillingDateNotifier(NotificationQueueService notificationQueueService, Bus eventBus, InvoiceConfig config){
+ this.notificationQueueService = notificationQueueService;
+ this.config = config;
+ this.eventBus = eventBus;
+ }
+
+
+ @LifecycleHandlerType(LifecycleLevel.INIT_SERVICE)
+ public void initialize() {
+ try {
+ nextBillingQueue = notificationQueueService.createNotificationQueue(NEXT_BILLING_DATE_NOTIFIER_SERVICE_NAME,
+ NEXT_BILLING_DATE_NOTIFIER_QUEUE,
+ new NotificationQueueHandler() {
+ @Override
+ public void handleReadyNotification(String notificationKey) {
+ UUID subscriptionId;
+ try {
+ subscriptionId = UUID.fromString(notificationKey);
+ } catch (IllegalArgumentException e) {
+ log.error("The key returned from the NextBillingNotificationQueue is not a valid UUID",e);
+ return;
+ }
+
+ processEvent(subscriptionId);
+ }
+ },
+ new NotificationConfig() {
+ @Override
+ public boolean isNotificationProcessingOff() {
+ return config.isEventProcessingOff();
+ }
+ @Override
+ public long getNotificationSleepTimeMs() {
+ return config.getNotificationSleepTimeMs();
+ }
+ @Override
+ public int getDaoMaxReadyEvents() {
+ return config.getDaoMaxReadyEvents();
+ }
+ @Override
+ public long getDaoClaimTimeMs() {
+ return config.getDaoMaxReadyEvents();
+ }
+ });
+ } catch (NotficationQueueAlreadyExists e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @LifecycleHandlerType(LifecycleLevel.START_SERVICE)
+ public void start() {
+ nextBillingQueue.startQueue();
+ }
+
+ @LifecycleHandlerType(LifecycleLevel.STOP_SERVICE)
+ public void stop() {
+ if (nextBillingQueue != null) {
+ nextBillingQueue.stopQueue();
+ }
+ }
+
+ @Override
+ public String getName() {
+ return NEXT_BILLING_DATE_NOTIFIER_SERVICE_NAME;
+ }
+
+ private void processEvent(UUID subscriptionId) {
+ try {
+ eventBus.post(new NextBillingDateEvent(subscriptionId));
+ } catch (EventBusException e) {
+ log.error("Failed to post entitlement event " + subscriptionId, e);
+ }
+ }
+
+ @Override
+ public void insertNextBillingNotification(Transmogrifier transactionalDao, final UUID subscriptionId, DateTime futureNotificationTime) {
+ nextBillingQueue.recordFutureNotificationFromTransaction(transactionalDao, futureNotificationTime, new NotificationKey(){
+ public String toString() {
+ return subscriptionId.toString();
+ }
+ });
+ }
+
+}
diff --git a/invoice/src/test/java/com/ning/billing/invoice/notification/TestNextBillingDateNotifier.java b/invoice/src/test/java/com/ning/billing/invoice/notification/TestNextBillingDateNotifier.java
new file mode 100644
index 0000000..c9f5aa9
--- /dev/null
+++ b/invoice/src/test/java/com/ning/billing/invoice/notification/TestNextBillingDateNotifier.java
@@ -0,0 +1,65 @@
+/*
+ * Copyright 2010-2011 Ning, Inc.
+ *
+ * Ning licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.ning.billing.invoice.notification;
+
+import org.skife.config.ConfigurationObjectFactory;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Stage;
+import com.ning.billing.config.CatalogConfig;
+import com.ning.billing.config.InvoiceConfig;
+import com.ning.billing.lifecycle.KillbillService.ServiceException;
+import com.ning.billing.util.clock.Clock;
+import com.ning.billing.util.clock.DefaultClock;
+import com.ning.billing.util.eventbus.Bus;
+import com.ning.billing.util.eventbus.MemoryEventBus;
+import com.ning.billing.util.notificationq.DefaultNotificationQueueService;
+import com.ning.billing.util.notificationq.NotificationQueueService;
+
+public class TestNextBillingDateNotifier {
+
+ private Clock clock;
+ private NextBillingDateNotifier notifier;
+
+ @BeforeClass(groups={"setup"})
+ public void setup() throws ServiceException {
+ //TestApiBase.loadSystemPropertiesFromClasspath("/entitlement.properties");
+ final Injector g = Guice.createInjector(Stage.PRODUCTION, new AbstractModule() {
+ protected void configure() {
+ bind(Clock.class).to(DefaultClock.class).asEagerSingleton();
+ bind(NextBillingDateNotifier.class).to(DefaultNextBillingDateNotifier.class).asEagerSingleton();
+ bind(Bus.class).to(MemoryEventBus.class).asEagerSingleton();
+ bind(NotificationQueueService.class).to(DefaultNotificationQueueService.class).asEagerSingleton();
+ final InvoiceConfig config = new ConfigurationObjectFactory(System.getProperties()).build(InvoiceConfig.class);
+ bind(InvoiceConfig.class).toInstance(config);
+ }
+ });
+
+ notifier = g.getInstance(NextBillingDateNotifier.class);
+ clock = g.getInstance(Clock.class);
+
+ }
+
+ @Test(enabled=false, groups="fast")
+ public void test() {
+
+ }
+}
