@Test public void putInflight() throws Exception { Injector injector = getInjector(); 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.DEFAULT, queueName, actorSystemFig.getRegionLocal(), null, System.currentTimeMillis(), null, queueMessageId ); qms.writeMessage( message ); QueueActorHelper helper = injector.getInstance( QueueActorHelper.class ); helper.putInflight( message ); Assert.assertNull( qms.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.DEFAULT, message.getQueueMessageId() ) ); Assert.assertNotNull( qms.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, DatabaseQueueMessage.Type.INFLIGHT, message.getQueueMessageId() ) ); 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.GET, auditLogs.getEntities().get( 0 ).getAction() ); DistributedQueueService distributedQueueService = injector.getInstance( DistributedQueueService.class ); distributedQueueService.shutdown(); } finally { queueManager.deleteQueue( queueName ); } }
boolean putInflight( DatabaseQueueMessage queueMessage ) { UUID qmid = queueMessage.getQueueMessageId(); try { messageSerialization.putInflight( queueMessage ); } catch ( Throwable t ) { logger.error("Error putting inflight queue message " + qmid + " queue name: " + queueMessage.getQueueName(), t); auditLogSerialization.recordAuditLog( AuditLog.Action.GET, AuditLog.Status.ERROR, queueMessage.getQueueName(), actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), qmid); return false; } auditLogSerialization.recordAuditLog( AuditLog.Action.GET, AuditLog.Status.SUCCESS, queueMessage.getQueueName(), actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), qmid); return true; }
QueueActorHelper { boolean putInflight( DatabaseQueueMessage queueMessage ) { UUID qmid = queueMessage.getQueueMessageId(); try { messageSerialization.putInflight( queueMessage ); } catch ( Throwable t ) { logger.error("Error putting inflight queue message " + qmid + " queue name: " + queueMessage.getQueueName(), t); auditLogSerialization.recordAuditLog( AuditLog.Action.GET, AuditLog.Status.ERROR, queueMessage.getQueueName(), actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), qmid); return false; } auditLogSerialization.recordAuditLog( AuditLog.Action.GET, AuditLog.Status.SUCCESS, queueMessage.getQueueName(), actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), qmid); return true; } }
QueueActorHelper { boolean putInflight( DatabaseQueueMessage queueMessage ) { UUID qmid = queueMessage.getQueueMessageId(); try { messageSerialization.putInflight( queueMessage ); } catch ( Throwable t ) { logger.error("Error putting inflight queue message " + qmid + " queue name: " + queueMessage.getQueueName(), t); auditLogSerialization.recordAuditLog( AuditLog.Action.GET, AuditLog.Status.ERROR, queueMessage.getQueueName(), actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), qmid); return false; } auditLogSerialization.recordAuditLog( AuditLog.Action.GET, AuditLog.Status.SUCCESS, queueMessage.getQueueName(), actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), qmid); return true; } @Inject  QueueActorHelper(
            QakkaFig                  qakkaFig,
            ActorSystemFig            actorSystemFig,
            QueueMessageSerialization messageSerialization,
            AuditLogSerialization     auditLogSerialization,
            InMemoryQueue             inMemoryQueue,
            MetricsService            metricsService,
            CassandraClient           cassandraClient
            ); }
QueueActorHelper { boolean putInflight( DatabaseQueueMessage queueMessage ) { UUID qmid = queueMessage.getQueueMessageId(); try { messageSerialization.putInflight( queueMessage ); } catch ( Throwable t ) { logger.error("Error putting inflight queue message " + qmid + " queue name: " + queueMessage.getQueueName(), t); auditLogSerialization.recordAuditLog( AuditLog.Action.GET, AuditLog.Status.ERROR, queueMessage.getQueueName(), actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), qmid); return false; } auditLogSerialization.recordAuditLog( AuditLog.Action.GET, AuditLog.Status.SUCCESS, queueMessage.getQueueName(), actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), qmid); return true; } @Inject  QueueActorHelper(
            QakkaFig                  qakkaFig,
            ActorSystemFig            actorSystemFig,
            QueueMessageSerialization messageSerialization,
            AuditLogSerialization     auditLogSerialization,
            InMemoryQueue             inMemoryQueue,
            MetricsService            metricsService,
            CassandraClient           cassandraClient
            );  }
QueueActorHelper { boolean putInflight( DatabaseQueueMessage queueMessage ) { UUID qmid = queueMessage.getQueueMessageId(); try { messageSerialization.putInflight( queueMessage ); } catch ( Throwable t ) { logger.error("Error putting inflight queue message " + qmid + " queue name: " + queueMessage.getQueueName(), t); auditLogSerialization.recordAuditLog( AuditLog.Action.GET, AuditLog.Status.ERROR, queueMessage.getQueueName(), actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), qmid); return false; } auditLogSerialization.recordAuditLog( AuditLog.Action.GET, AuditLog.Status.SUCCESS, queueMessage.getQueueName(), actorSystemFig.getRegionLocal(), queueMessage.getMessageId(), qmid); return true; } @Inject  QueueActorHelper(
            QakkaFig                  qakkaFig,
            ActorSystemFig            actorSystemFig,
            QueueMessageSerialization messageSerialization,
            AuditLogSerialization     auditLogSerialization,
            InMemoryQueue             inMemoryQueue,
            MetricsService            metricsService,
            CassandraClient           cassandraClient
            );   }