@Test public void ackQueueMessage() throws Exception { Injector injector = getInjector(); CassandraClient cassandraClient = injector.getInstance( CassandraClientImpl.class ); injector.getInstance( App.class ); ActorSystemFig actorSystemFig = injector.getInstance( ActorSystemFig.class ); QueueMessageSerialization qms = injector.getInstance( QueueMessageSerialization.class ); QueueManager queueManager = injector.getInstance( QueueManager.class ); String region = actorSystemFig.getRegionLocal(); UUID queueMessageId = QakkaUtils.getTimeUuid(); String queueName = "qat_queue_" + RandomStringUtils.randomAlphanumeric( 10 ); queueManager.createQueue( new Queue( queueName ) ); try { DatabaseQueueMessage message = new DatabaseQueueMessage( QakkaUtils.getTimeUuid(), DatabaseQueueMessage.Type.INFLIGHT, queueName, actorSystemFig.getRegionLocal(), null, System.currentTimeMillis(), null, queueMessageId ); qms.writeMessage( message ); QueueActorHelper helper = injector.getInstance( QueueActorHelper.class ); helper.ackQueueMessage( queueName, message.getQueueMessageId() ); Assert.assertNull( helper.loadDatabaseQueueMessage( queueName, QakkaUtils.getTimeUuid(), DatabaseQueueMessage.Type.INFLIGHT ) ); Assert.assertNull( helper.loadDatabaseQueueMessage( queueName, QakkaUtils.getTimeUuid(), DatabaseQueueMessage.Type.DEFAULT ) ); AuditLogSerialization auditLogSerialization = injector.getInstance( AuditLogSerialization.class ); Result<AuditLog> auditLogs = auditLogSerialization.getAuditLogs( message.getMessageId() ); Assert.assertEquals( 1, auditLogs.getEntities().size() ); Assert.assertEquals( AuditLog.Status.SUCCESS, auditLogs.getEntities().get( 0 ).getStatus() ); Assert.assertEquals( AuditLog.Action.ACK, auditLogs.getEntities().get( 0 ).getAction() ); DistributedQueueService distributedQueueService = injector.getInstance( DistributedQueueService.class ); distributedQueueService.shutdown(); } finally { queueManager.deleteQueue( queueName ); } }
DistributedQueueService.Status ackQueueMessage(String queueName, UUID queueMessageId ) { DatabaseQueueMessage queueMessage = messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); if ( queueMessage == null ) { logger.error("Queue {} queue message id {} not found in inflight table", queueName, queueMessageId); return DistributedQueueService.Status.NOT_INFLIGHT; } boolean error = false; try { messageSerialization.deleteMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); } catch (Throwable t) { logger.error( "Error deleting queueMessage for ack", t ); error = true; } if ( !error ) { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.SUCCESS, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.SUCCESS; } else { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.ERROR, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.ERROR; } }
QueueActorHelper { DistributedQueueService.Status ackQueueMessage(String queueName, UUID queueMessageId ) { DatabaseQueueMessage queueMessage = messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); if ( queueMessage == null ) { logger.error("Queue {} queue message id {} not found in inflight table", queueName, queueMessageId); return DistributedQueueService.Status.NOT_INFLIGHT; } boolean error = false; try { messageSerialization.deleteMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); } catch (Throwable t) { logger.error( "Error deleting queueMessage for ack", t ); error = true; } if ( !error ) { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.SUCCESS, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.SUCCESS; } else { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.ERROR, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.ERROR; } } }
QueueActorHelper { DistributedQueueService.Status ackQueueMessage(String queueName, UUID queueMessageId ) { DatabaseQueueMessage queueMessage = messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); if ( queueMessage == null ) { logger.error("Queue {} queue message id {} not found in inflight table", queueName, queueMessageId); return DistributedQueueService.Status.NOT_INFLIGHT; } boolean error = false; try { messageSerialization.deleteMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); } catch (Throwable t) { logger.error( "Error deleting queueMessage for ack", t ); error = true; } if ( !error ) { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.SUCCESS, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.SUCCESS; } else { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.ERROR, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.ERROR; } } @Inject  QueueActorHelper(
            QakkaFig                  qakkaFig,
            ActorSystemFig            actorSystemFig,
            QueueMessageSerialization messageSerialization,
            AuditLogSerialization     auditLogSerialization,
            InMemoryQueue             inMemoryQueue,
            MetricsService            metricsService,
            CassandraClient           cassandraClient
            ); }
QueueActorHelper { DistributedQueueService.Status ackQueueMessage(String queueName, UUID queueMessageId ) { DatabaseQueueMessage queueMessage = messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); if ( queueMessage == null ) { logger.error("Queue {} queue message id {} not found in inflight table", queueName, queueMessageId); return DistributedQueueService.Status.NOT_INFLIGHT; } boolean error = false; try { messageSerialization.deleteMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); } catch (Throwable t) { logger.error( "Error deleting queueMessage for ack", t ); error = true; } if ( !error ) { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.SUCCESS, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.SUCCESS; } else { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.ERROR, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.ERROR; } } @Inject  QueueActorHelper(
            QakkaFig                  qakkaFig,
            ActorSystemFig            actorSystemFig,
            QueueMessageSerialization messageSerialization,
            AuditLogSerialization     auditLogSerialization,
            InMemoryQueue             inMemoryQueue,
            MetricsService            metricsService,
            CassandraClient           cassandraClient
            );  }
QueueActorHelper { DistributedQueueService.Status ackQueueMessage(String queueName, UUID queueMessageId ) { DatabaseQueueMessage queueMessage = messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); if ( queueMessage == null ) { logger.error("Queue {} queue message id {} not found in inflight table", queueName, queueMessageId); return DistributedQueueService.Status.NOT_INFLIGHT; } boolean error = false; try { messageSerialization.deleteMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, queueMessageId ); } catch (Throwable t) { logger.error( "Error deleting queueMessage for ack", t ); error = true; } if ( !error ) { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.SUCCESS, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.SUCCESS; } else { auditLogSerialization.recordAuditLog( AuditLog.Action.ACK, AuditLog.Status.ERROR, queueName, actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), queueMessageId ); return DistributedQueueService.Status.ERROR; } } @Inject  QueueActorHelper(
            QakkaFig                  qakkaFig,
            ActorSystemFig            actorSystemFig,
            QueueMessageSerialization messageSerialization,
            AuditLogSerialization     auditLogSerialization,
            InMemoryQueue             inMemoryQueue,
            MetricsService            metricsService,
            CassandraClient           cassandraClient
            );   }