@Test public void loadDatabaseQueueMessage() 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(); String queueName = "qat_queue_" + RandomStringUtils.randomAlphanumeric( 10 ); try { queueManager.createQueue( new Queue( queueName ) ); UUID queueMessageId = QakkaUtils.getTimeUuid(); 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 ); DatabaseQueueMessage queueMessage = helper.loadDatabaseQueueMessage( queueName, message.getQueueMessageId(), message.getType() ); Assert.assertNotNull( queueMessage ); DistributedQueueService distributedQueueService = injector.getInstance( DistributedQueueService.class ); distributedQueueService.shutdown(); } finally { queueManager.deleteQueue( queueName ); } }
DatabaseQueueMessage loadDatabaseQueueMessage( String queueName, UUID queueMessageId, DatabaseQueueMessage.Type type ) { try { return messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, type, queueMessageId ); } catch (Throwable t) { logger.error( "Error reading queueMessage", t ); } return null; }
QueueActorHelper { DatabaseQueueMessage loadDatabaseQueueMessage( String queueName, UUID queueMessageId, DatabaseQueueMessage.Type type ) { try { return messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, type, queueMessageId ); } catch (Throwable t) { logger.error( "Error reading queueMessage", t ); } return null; } }
QueueActorHelper { DatabaseQueueMessage loadDatabaseQueueMessage( String queueName, UUID queueMessageId, DatabaseQueueMessage.Type type ) { try { return messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, type, queueMessageId ); } catch (Throwable t) { logger.error( "Error reading queueMessage", t ); } return null; } @Inject  QueueActorHelper(
            QakkaFig                  qakkaFig,
            ActorSystemFig            actorSystemFig,
            QueueMessageSerialization messageSerialization,
            AuditLogSerialization     auditLogSerialization,
            InMemoryQueue             inMemoryQueue,
            MetricsService            metricsService,
            CassandraClient           cassandraClient
            ); }
QueueActorHelper { DatabaseQueueMessage loadDatabaseQueueMessage( String queueName, UUID queueMessageId, DatabaseQueueMessage.Type type ) { try { return messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, type, queueMessageId ); } catch (Throwable t) { logger.error( "Error reading queueMessage", t ); } return null; } @Inject  QueueActorHelper(
            QakkaFig                  qakkaFig,
            ActorSystemFig            actorSystemFig,
            QueueMessageSerialization messageSerialization,
            AuditLogSerialization     auditLogSerialization,
            InMemoryQueue             inMemoryQueue,
            MetricsService            metricsService,
            CassandraClient           cassandraClient
            );  }
QueueActorHelper { DatabaseQueueMessage loadDatabaseQueueMessage( String queueName, UUID queueMessageId, DatabaseQueueMessage.Type type ) { try { return messageSerialization.loadMessage( queueName, actorSystemFig.getRegionLocal(), null, type, queueMessageId ); } catch (Throwable t) { logger.error( "Error reading queueMessage", t ); } return null; } @Inject  QueueActorHelper(
            QakkaFig                  qakkaFig,
            ActorSystemFig            actorSystemFig,
            QueueMessageSerialization messageSerialization,
            AuditLogSerialization     auditLogSerialization,
            InMemoryQueue             inMemoryQueue,
            MetricsService            metricsService,
            CassandraClient           cassandraClient
            );   }