Details
diff --git a/account/src/main/java/com/ning/billing/account/api/svcs/DefaultAccountInternalApi.java b/account/src/main/java/com/ning/billing/account/api/svcs/DefaultAccountInternalApi.java
index f0189dd..b29bc91 100644
--- a/account/src/main/java/com/ning/billing/account/api/svcs/DefaultAccountInternalApi.java
+++ b/account/src/main/java/com/ning/billing/account/api/svcs/DefaultAccountInternalApi.java
@@ -46,7 +46,7 @@ public class DefaultAccountInternalApi implements AccountInternalApi {
}
@Override
- public Account getAccountById(UUID accountId, InternalTenantContext context)
+ public Account getAccountById(final UUID accountId, final InternalTenantContext context)
throws AccountApiException {
final Account account = accountDao.getById(accountId, context);
if (account == null) {
@@ -56,25 +56,31 @@ public class DefaultAccountInternalApi implements AccountInternalApi {
}
@Override
- public void updateAccount(String externalKey, AccountData accountData,
- InternalCallContext context) throws AccountApiException {
+ public Account getAccountByRecordId(final Long recordId,
+ final InternalTenantContext context) throws AccountApiException {
+ return accountDao.getByRecordId(recordId, context);
+ }
+
+ @Override
+ public void updateAccount(final String externalKey, final AccountData accountData,
+ final InternalCallContext context) throws AccountApiException {
final Account account = getAccountByKey(externalKey, context);
try {
final Account updatedAccount = new DefaultAccount(account.getId(), accountData);
accountDao.update(updatedAccount,context);
- } catch (EntityPersistenceException e) {
+ } catch (final EntityPersistenceException e) {
throw new AccountApiException(e, ErrorCode.ACCOUNT_UPDATE_FAILED);
}
}
@Override
- public List<AccountEmail> getEmails(UUID accountId,
- InternalTenantContext context) {
+ public List<AccountEmail> getEmails(final UUID accountId,
+ final InternalTenantContext context) {
return accountEmailDao.getEmails(accountId, context);
}
@Override
- public Account getAccountByKey(String key, InternalTenantContext context)
+ public Account getAccountByKey(final String key, final InternalTenantContext context)
throws AccountApiException {
final Account account = accountDao.getAccountByKey(key, context);
if (account == null) {
@@ -84,17 +90,17 @@ public class DefaultAccountInternalApi implements AccountInternalApi {
}
@Override
- public void removePaymentMethod(UUID accountId, InternalCallContext context)
+ public void removePaymentMethod(final UUID accountId, final InternalCallContext context)
throws AccountApiException {
updatePaymentMethod(accountId, null, context);
}
@Override
- public void updatePaymentMethod(UUID accountId, UUID paymentMethodId,
- InternalCallContext context) throws AccountApiException {
+ public void updatePaymentMethod(final UUID accountId, final UUID paymentMethodId,
+ final InternalCallContext context) throws AccountApiException {
try {
accountDao.updatePaymentMethod(accountId, paymentMethodId, context);
- } catch (EntityPersistenceException e) {
+ } catch (final EntityPersistenceException e) {
throw new AccountApiException(e, e.getCode(), e.getMessage());
}
}
diff --git a/account/src/main/java/com/ning/billing/account/dao/AuditedAccountDao.java b/account/src/main/java/com/ning/billing/account/dao/AuditedAccountDao.java
index 808a78d..33da659 100644
--- a/account/src/main/java/com/ning/billing/account/dao/AuditedAccountDao.java
+++ b/account/src/main/java/com/ning/billing/account/dao/AuditedAccountDao.java
@@ -80,6 +80,12 @@ public class AuditedAccountDao implements AccountDao {
}
@Override
+ public Account getByRecordId(final Long recordId, final InternalTenantContext context) {
+ return accountSqlDao.getByRecordId(recordId, context);
+ }
+
+
+ @Override
public List<Account> get(final InternalTenantContext context) {
return accountSqlDao.get(context);
}
@@ -122,13 +128,13 @@ public class AuditedAccountDao implements AccountDao {
context.getTenantRecordId());
try {
eventBus.postFromTransaction(creationEvent, transactionalDao, rehydratedContext);
- } catch (EventBusException e) {
+ } catch (final EventBusException e) {
log.warn("Failed to post account creation event for account " + account.getId(), e);
}
return null;
}
});
- } catch (RuntimeException re) {
+ } catch (final RuntimeException re) {
if (re.getCause() instanceof EntityPersistenceException) {
throw (EntityPersistenceException) re.getCause();
} else if (re.getCause() instanceof DataTruncation) {
@@ -173,14 +179,14 @@ public class AuditedAccountDao implements AccountDao {
if (changeEvent.hasChanges()) {
try {
eventBus.postFromTransaction(changeEvent, transactional, context);
- } catch (EventBusException e) {
+ } catch (final EventBusException e) {
log.warn("Failed to post account change event for account " + accountId, e);
}
}
return null;
}
});
- } catch (RuntimeException re) {
+ } catch (final RuntimeException re) {
if (re.getCause() instanceof EntityPersistenceException) {
throw (EntityPersistenceException) re.getCause();
} else {
@@ -224,14 +230,14 @@ public class AuditedAccountDao implements AccountDao {
if (changeEvent.hasChanges()) {
try {
eventBus.postFromTransaction(changeEvent, transactional, context);
- } catch (EventBusException e) {
+ } catch (final EventBusException e) {
log.warn("Failed to post account change event for account " + accountId, e);
}
}
return null;
}
});
- } catch (RuntimeException re) {
+ } catch (final RuntimeException re) {
if (re.getCause() instanceof EntityPersistenceException) {
throw (EntityPersistenceException) re.getCause();
} else {
diff --git a/account/src/main/java/com/ning/billing/account/dao/AuditedAccountEmailDao.java b/account/src/main/java/com/ning/billing/account/dao/AuditedAccountEmailDao.java
index a3428f4..c161b31 100644
--- a/account/src/main/java/com/ning/billing/account/dao/AuditedAccountEmailDao.java
+++ b/account/src/main/java/com/ning/billing/account/dao/AuditedAccountEmailDao.java
@@ -65,6 +65,12 @@ public class AuditedAccountEmailDao extends AuditedCollectionDaoBase<AccountEmai
}
@Override
+ public AccountEmail getByRecordId(final Long recordId, final InternalTenantContext context) {
+ throw new UnsupportedOperationException();
+ }
+
+
+ @Override
public List<AccountEmail> get(final InternalTenantContext context) {
throw new UnsupportedOperationException();
}
diff --git a/account/src/main/resources/com/ning/billing/account/dao/AccountSqlDao.sql.stg b/account/src/main/resources/com/ning/billing/account/dao/AccountSqlDao.sql.stg
index 45e945f..0329a2f 100644
--- a/account/src/main/resources/com/ning/billing/account/dao/AccountSqlDao.sql.stg
+++ b/account/src/main/resources/com/ning/billing/account/dao/AccountSqlDao.sql.stg
@@ -98,6 +98,12 @@ getRecordId() ::= <<
WHERE id = :id <AND_CHECK_TENANT()>;
>>
+getByRecordId() ::= <<
+ SELECT <accountFields()>
+ FROM accounts
+ WHERE record_id = :recordId <AND_CHECK_TENANT()>;
+>>
+
getHistoryRecordId() ::= <<
SELECT MAX(history_record_id)
FROM account_history
diff --git a/account/src/test/java/com/ning/billing/account/dao/MockAccountDao.java b/account/src/test/java/com/ning/billing/account/dao/MockAccountDao.java
index 53d3f2e..fddfbc1 100644
--- a/account/src/test/java/com/ning/billing/account/dao/MockAccountDao.java
+++ b/account/src/test/java/com/ning/billing/account/dao/MockAccountDao.java
@@ -55,7 +55,7 @@ public class MockAccountDao implements AccountDao {
try {
eventBus.post(new DefaultAccountCreationEvent(account, null, 1L, 1L), context);
- } catch (EventBusException ex) {
+ } catch (final EventBusException ex) {
throw new RuntimeException(ex);
}
}
@@ -98,7 +98,7 @@ public class MockAccountDao implements AccountDao {
if (changeEvent.hasChanges()) {
try {
eventBus.post(changeEvent, context);
- } catch (EventBusException ex) {
+ } catch (final EventBusException ex) {
throw new RuntimeException(ex);
}
}
@@ -107,4 +107,9 @@ public class MockAccountDao implements AccountDao {
@Override
public void updatePaymentMethod(final UUID accountId, final UUID paymentMethodId, final InternalCallContext context) throws EntityPersistenceException {
}
+
+ @Override
+ public Account getByRecordId(final Long recordId, final InternalTenantContext context) {
+ return null;
+ }
}
diff --git a/account/src/test/java/com/ning/billing/account/dao/MockAccountEmailDao.java b/account/src/test/java/com/ning/billing/account/dao/MockAccountEmailDao.java
index e6951ae..8139bf0 100644
--- a/account/src/test/java/com/ning/billing/account/dao/MockAccountEmailDao.java
+++ b/account/src/test/java/com/ning/billing/account/dao/MockAccountEmailDao.java
@@ -94,4 +94,10 @@ public class MockAccountEmailDao implements AccountEmailDao {
@Override
public void test(final InternalTenantContext context) {
}
+
+ @Override
+ public AccountEmail getByRecordId(final Long recordId,
+ final InternalTenantContext context) {
+ return null;
+ }
}
diff --git a/beatrix/src/main/java/com/ning/billing/beatrix/extbus/PersistentExternalBus.java b/beatrix/src/main/java/com/ning/billing/beatrix/extbus/PersistentExternalBus.java
index 6a37476..81c832a 100644
--- a/beatrix/src/main/java/com/ning/billing/beatrix/extbus/PersistentExternalBus.java
+++ b/beatrix/src/main/java/com/ning/billing/beatrix/extbus/PersistentExternalBus.java
@@ -27,6 +27,8 @@ import org.skife.jdbi.v2.IDBI;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.ning.billing.account.api.Account;
+import com.ning.billing.account.api.AccountApiException;
import com.ning.billing.beatrix.bus.api.ExtBusEvent;
import com.ning.billing.beatrix.bus.api.ExternalBus;
import com.ning.billing.beatrix.extbus.dao.ExtBusEventEntry;
@@ -40,6 +42,7 @@ import com.ning.billing.util.callcontext.InternalCallContextFactory;
import com.ning.billing.util.callcontext.UserType;
import com.ning.billing.util.clock.Clock;
import com.ning.billing.util.queue.PersistentQueueBase;
+import com.ning.billing.util.svcapi.account.AccountInternalApi;
import com.ning.billing.util.svcsapi.bus.InternalBus.EventBusException;
import com.google.common.eventbus.EventBus;
@@ -58,6 +61,7 @@ public class PersistentExternalBus extends PersistentQueueBase implements Extern
private final Clock clock;
private final String hostname;
private final InternalCallContextFactory internalCallContextFactory;
+ private final AccountInternalApi accountApi;
private static final class EventBusDelegate extends EventBus {
@@ -67,7 +71,7 @@ public class PersistentExternalBus extends PersistentQueueBase implements Extern
}
@Inject
- public PersistentExternalBus(final IDBI dbi, final Clock clock, final PersistentBusConfig config, final InternalCallContextFactory internalCallContextFactory) {
+ public PersistentExternalBus(final AccountInternalApi accountApi, final IDBI dbi, final Clock clock, final PersistentBusConfig config, final InternalCallContextFactory internalCallContextFactory) {
super("Bus", Executors.newFixedThreadPool(config.getNbThreads(), new ThreadFactory() {
@Override
public Thread newThread(final Runnable r) {
@@ -81,6 +85,7 @@ public class PersistentExternalBus extends PersistentQueueBase implements Extern
this.eventBusDelegate = new EventBusDelegate("Killbill EventBus");
this.hostname = Hostname.get();
this.internalCallContextFactory = internalCallContextFactory;
+ this.accountApi = accountApi;
}
public void start() {
@@ -103,11 +108,8 @@ public class PersistentExternalBus extends PersistentQueueBase implements Extern
int result = 0;
for (final ExtBusEventEntry cur : events) {
-
- // API_FIX How do we get UUID from recordId ?
- final UUID accountId = null;
- final UUID tenantId = null;
- final ExtBusEvent event = new DefaultBusEvent(cur.getExtBusType(), cur.getObjectType(), cur.getObjectId(), accountId, tenantId);
+ final UUID accountId = getAccountIdFromRecordId(cur.getAccountRecordId(), context);
+ final ExtBusEvent event = new DefaultBusEvent(cur.getExtBusType(), cur.getObjectType(), cur.getObjectId(), accountId, null);
result++;
// STEPH exception handling is done by GUAVA-- logged a bug Issue-780
eventBusDelegate.post(event);
@@ -117,6 +119,16 @@ public class PersistentExternalBus extends PersistentQueueBase implements Extern
return result;
}
+ private final UUID getAccountIdFromRecordId(final Long recordId, final InternalCallContext context) {
+ try {
+ final Account account = accountApi.getAccountByRecordId(recordId, context);
+ return account.getId();
+ } catch (final AccountApiException e) {
+ log.warn("Failed to retrieve acount from recordId {}", recordId);
+ return null;
+ }
+ }
+
private List<ExtBusEventEntry> getNextBusEvent(final InternalCallContext context) {
final Date now = clock.getUTCNow().toDate();
final Date nextAvailable = clock.getUTCNow().plus(DELTA_IN_PROCESSING_TIME_MS).toDate();
diff --git a/jaxrs/src/main/java/com/ning/billing/jaxrs/json/NotificationJson.java b/jaxrs/src/main/java/com/ning/billing/jaxrs/json/NotificationJson.java
index 9ddb2b8..762a184 100644
--- a/jaxrs/src/main/java/com/ning/billing/jaxrs/json/NotificationJson.java
+++ b/jaxrs/src/main/java/com/ning/billing/jaxrs/json/NotificationJson.java
@@ -44,8 +44,7 @@ public class NotificationJson {
public NotificationJson(final ExtBusEvent event) {
- // API_FIX accountId NULL
- this(event.getEventType().toString(), null /* event.getAccountId().toString() */, event.getObjectType().toString(), event.getObjectId().toString());
+ this(event.getEventType().toString(), event.getAccountId().toString(), event.getObjectType().toString(), event.getObjectId().toString());
}
diff --git a/server/src/test/java/com/ning/billing/jaxrs/KillbillClient.java b/server/src/test/java/com/ning/billing/jaxrs/KillbillClient.java
index 9062bf8..9e3fb83 100644
--- a/server/src/test/java/com/ning/billing/jaxrs/KillbillClient.java
+++ b/server/src/test/java/com/ning/billing/jaxrs/KillbillClient.java
@@ -163,6 +163,18 @@ public abstract class KillbillClient extends ServerTestSuiteWithEmbeddedDB {
// ACCOUNT UTILITIES
//
+ protected AccountJson getAccountById(final String id) throws Exception {
+ final String uri = JaxrsResource.ACCOUNTS_PATH + "/" + id;
+ final Response response = doGet(uri, DEFAULT_EMPTY_QUERY, DEFAULT_HTTP_TIMEOUT_SEC);
+ Assert.assertEquals(response.getStatusCode(), Status.OK.getStatusCode());
+
+ final String baseJson = response.getResponseBody();
+ final AccountJson objFromJson = mapper.readValue(baseJson, AccountJson.class);
+ Assert.assertNotNull(objFromJson);
+
+ return objFromJson;
+ }
+
protected AccountJson getAccountByExternalKey(final String externalKey) throws Exception {
final Response response = getAccountByExternalKeyNoValidation(externalKey);
Assert.assertEquals(response.getStatusCode(), Status.OK.getStatusCode());
diff --git a/server/src/test/java/com/ning/billing/jaxrs/TestPushNotification.java b/server/src/test/java/com/ning/billing/jaxrs/TestPushNotification.java
index b6b6704..94946af 100644
--- a/server/src/test/java/com/ning/billing/jaxrs/TestPushNotification.java
+++ b/server/src/test/java/com/ning/billing/jaxrs/TestPushNotification.java
@@ -48,22 +48,20 @@ public class TestPushNotification extends TestJaxrsBase {
private final static String CALLBACK_ENDPPOINT = "/callmeback";
private volatile boolean callbackCompleted;
+ private volatile boolean callbackCompletedWithError;
@BeforeMethod(groups = "slow")
public void startServer() throws Exception {
callbackServer = new CallbackServer(this, SERVER_PORT, CALLBACK_ENDPPOINT);
callbackCompleted = false;
+ callbackCompletedWithError = false;
callbackServer.startServer();
}
@AfterMethod(groups = "slow")
public void stopServer() throws Exception {
- final boolean success = waitForCallbacksToComplete();
- callbackServer.stopServer();
- if (!success) {
- Assert.fail("Fail to see push notification callbacks after 5 sec");
- }
+ callbackServer.stopServer();
}
private boolean waitForCallbacksToComplete() throws InterruptedException {
@@ -79,16 +77,35 @@ public class TestPushNotification extends TestJaxrsBase {
return (remainingMs > 0);
}
+ public void retrieveAccountWithAsserts(final String accountId) {
+ try {
+ // Just check we can retrieve the account with the id from the callback
+ /* final AccountJson account = */ getAccountById(accountId);
+ } catch(final Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ }
+
@Test(groups = "slow")
- public void testTenant() throws Exception {
+ public void testPushNotification() throws Exception {
// Register tenant for callback
registerCallbackNotificationForTenant("http://127.0.0.1:" + SERVER_PORT + CALLBACK_ENDPPOINT);
// Create account to trigger a push notification
createAccount();
+
+ final boolean success = waitForCallbacksToComplete();
+ if (!success) {
+ Assert.fail("Fail to see push notification callbacks after 5 sec");
+ }
+
+ if (callbackCompletedWithError) {
+ Assert.fail("Assertion during callback failed...");
+ }
}
- public void setCompleted() {
+ public void setCompleted(final boolean withError) {
callbackCompleted = true;
+ callbackCompletedWithError = withError;
}
public static class CallbackServer {
@@ -128,10 +145,13 @@ public class TestPushNotification extends TestJaxrsBase {
private final TestPushNotification test;
private final ObjectMapper objectMapper = new ObjectMapper();
+ private boolean withError;
+
public CallmebackServlet(final TestPushNotification test, final int expectedNbCalls) {
this.expectedNbCalls = expectedNbCalls;
this.test = test;
this.receivedCalls = new AtomicInteger(0);
+ this.withError = false;
}
@Override
@@ -140,26 +160,33 @@ public class TestPushNotification extends TestJaxrsBase {
final String body = CharStreams.toString( new InputStreamReader(request.getInputStream(), "UTF-8" ));
+ response.setContentType("application/json");
+ response.setStatus(HttpServletResponse.SC_OK);
+
log.info("Got body {}", body);
- final NotificationJson notification = objectMapper.readValue(body, NotificationJson.class);
- Assert.assertEquals(notification.getEventType(), "ACCOUNT_CREATION");
- Assert.assertEquals(notification.getObjectType(), "ACCOUNT");
- // Not checking the ID returned we we should
+ try {
+ final NotificationJson notification = objectMapper.readValue(body, NotificationJson.class);
+ Assert.assertEquals(notification.getEventType(), "ACCOUNT_CREATION");
+ Assert.assertEquals(notification.getObjectType(), "ACCOUNT");
+ Assert.assertNotNull(notification.getObjectId());
+ Assert.assertNotNull(notification.getAccountId());
+ Assert.assertEquals(notification.getObjectId(), notification.getAccountId());
+
+ test.retrieveAccountWithAsserts(notification.getObjectId());
+ } catch (final AssertionError e) {
+ withError = true;
+ }
log.info("CallmebackServlet received {} calls , current = {}", current, body);
-
- response.setContentType("application/json");
- response.setStatus(HttpServletResponse.SC_OK);
-
- stopServerWhenComplete(current);
+ stopServerWhenComplete(current, withError);
}
- private void stopServerWhenComplete(final int current) {
+ private void stopServerWhenComplete(final int current, final boolean withError) {
if (current == expectedNbCalls) {
log.info("Excellent, we are done!");
- test.setCompleted();
+ test.setCompleted(withError);
}
}
}
diff --git a/tenant/src/main/java/com/ning/billing/tenant/dao/DefaultTenantDao.java b/tenant/src/main/java/com/ning/billing/tenant/dao/DefaultTenantDao.java
index 6f59f17..16620e3 100644
--- a/tenant/src/main/java/com/ning/billing/tenant/dao/DefaultTenantDao.java
+++ b/tenant/src/main/java/com/ning/billing/tenant/dao/DefaultTenantDao.java
@@ -122,4 +122,9 @@ public class DefaultTenantDao implements TenantDao {
public void deleteTenantKey(final String key, final InternalCallContext context) {
tenantKVSqlDao.deleteTenantKey(key, context.getTenantRecordId());
}
+
+ @Override
+ public Tenant getByRecordId(final Long recordId, final InternalTenantContext context) {
+ throw new UnsupportedOperationException();
+ }
}
diff --git a/util/src/main/java/com/ning/billing/util/entity/dao/EntityDao.java b/util/src/main/java/com/ning/billing/util/entity/dao/EntityDao.java
index 3d4ab80..370f0e9 100644
--- a/util/src/main/java/com/ning/billing/util/entity/dao/EntityDao.java
+++ b/util/src/main/java/com/ning/billing/util/entity/dao/EntityDao.java
@@ -30,6 +30,8 @@ public interface EntityDao<T extends Entity> {
public Long getRecordId(UUID id, InternalTenantContext context);
+ public T getByRecordId(Long recordId, InternalTenantContext context);
+
public T getById(UUID id, InternalTenantContext context);
public List<T> get(InternalTenantContext context);
diff --git a/util/src/main/java/com/ning/billing/util/svcapi/account/AccountInternalApi.java b/util/src/main/java/com/ning/billing/util/svcapi/account/AccountInternalApi.java
index 006278f..60b36b1 100644
--- a/util/src/main/java/com/ning/billing/util/svcapi/account/AccountInternalApi.java
+++ b/util/src/main/java/com/ning/billing/util/svcapi/account/AccountInternalApi.java
@@ -31,6 +31,8 @@ public interface AccountInternalApi {
public Account getAccountById(UUID accountId, InternalTenantContext context) throws AccountApiException;
+ public Account getAccountByRecordId(Long recordId, InternalTenantContext context) throws AccountApiException;
+
public void updateAccount(String key, AccountData accountData, InternalCallContext context) throws AccountApiException;
public List<AccountEmail> getEmails(UUID accountId, InternalTenantContext context);
