MessageProcessor implements Predicate<Message> { @Override public boolean test(@Nullable final Message message) { if (message == null) { return false; } if (LOGGER.isTraceEnabled()) { LOGGER.trace("Received SQS message: {}", message); } else if (LOGGER.isDebugEnabled()) { LOGGER.debug("Received SQS message: {}", message.getMessageId()); } if (broadcaster.isEmpty()) { LOGGER.debug("No connections found, skipping SQS message processing"); return false; } final AmazonSNSNotification notification; try { notification = MAPPER.readValue(message.getBody(), AmazonSNSNotification.class); } catch (IOException e) { LOGGER.error("Failed to parse SNS notification, deleting SQS message", e); return true; } LOGGER.debug( "SNS notification created at: {} ({} behind)", notification.getTimestamp(), notification.getDelayDuration()); final String body; if (notification.isValid()) { body = notification.getMessage(); } else { body = message.getBody(); } final S3EventNotification records; try { records = S3EventNotification.parseJson(body); } catch (AmazonClientException e) { LOGGER.error("Failed to parse S3 event records, deleting SQS message", e); return true; } final int recordCount = records.getRecords().size(); recordCounts.update(recordCount); LOGGER.debug("Parsed {} S3 event records from SNS notification", recordCount); if (recordCount < 1) { LOGGER.debug("No S3 event records found in SNS notification, deleting SQS message"); return true; } int recordsProcessed = 0; for (S3EventNotificationRecord record : records.getRecords()) { if (broadcaster.isEmpty()) { LOGGER.debug("No connections found, not downloading from S3"); return false; } if (processRecord(record)) { recordsProcessed++; } } if (recordsProcessed == recordCount) { LOGGER.debug( "Processed {} of {} records, deleting SQS message", recordsProcessed, recordCount); return true; } LOGGER.debug( "Processed {} of {} records, not deleting SQS message: {}", recordsProcessed, recordCount, message.getMessageId()); return false; } MessageProcessor(
      @Nonnull final AmazonS3Downloader s3, @Nonnull final InstrumentedSseBroadcaster broadcaster); @Override boolean test(@Nullable final Message message);  }
@Test public void testProcessNullMessage() throws Exception { final boolean actual = processor.test(null); verify(broadcaster, never()).isEmpty(); verify(broadcaster, never()).broadcast(any(OutboundEvent.class)); verify(s3, never()).fetch(any(S3EventNotificationRecord.class)); assertThat(actual).isFalse(); }
@Test public void testProcessNoConnections() throws Exception { when(broadcaster.isEmpty()).thenReturn(true); final boolean actual = processor.test(message); verify(broadcaster).isEmpty(); verify(broadcaster, never()).test(anyString()); verify(s3, never()).fetch(any(S3EventNotificationRecord.class)); assertThat(actual).isFalse(); }
@Test public void testProcessSNSParseFailure() throws Exception { when(broadcaster.isEmpty()).thenReturn(false); final boolean actual = processor.test(message); verify(broadcaster).isEmpty(); verify(broadcaster, never()).test(anyString()); verify(s3, never()).fetch(any(S3EventNotificationRecord.class)); assertThat(actual).isTrue(); }
@Test public void testProcessS3EventFetchFailure() throws Exception { when(broadcaster.isEmpty()).thenReturn(false); when(s3.fetch(any(S3EventNotificationRecord.class))) .thenThrow(new AmazonServiceException("error")); message.setBody(FixtureHelpers.fixture("fixtures/sns_notification.json")); final boolean actual = processor.test(message); verify(broadcaster, times(2)).isEmpty(); verify(broadcaster, never()).test(anyString()); verify(s3).fetch(any(S3EventNotificationRecord.class)); assertThat(actual).isFalse(); }
@Test public void testProcessNoConnectionsAfterParse() throws Exception { when(broadcaster.isEmpty()).thenReturn(false, true); message.setBody(FixtureHelpers.fixture("fixtures/sns_notification.json")); final boolean actual = processor.test(message); verify(broadcaster, times(2)).isEmpty(); verify(broadcaster, never()).test(anyString()); verify(s3, never()).fetch(any(S3EventNotificationRecord.class)); assertThat(actual).isFalse(); }
@Test public void testProcessS3ZeroSizeFailure() throws Exception { when(broadcaster.isEmpty()).thenReturn(false); when(s3.fetch(any(S3EventNotificationRecord.class))).thenThrow(new AmazonS3ZeroSizeException()); message.setBody(FixtureHelpers.fixture("fixtures/sns_notification.json")); final boolean actual = processor.test(message); verify(broadcaster, times(2)).isEmpty(); verify(broadcaster, never()).test(anyString()); verify(s3).fetch(any(S3EventNotificationRecord.class)); assertThat(actual).isTrue(); }
@Test public void testProcessS3ConstraintFailure() throws Exception { when(broadcaster.isEmpty()).thenReturn(false); when(s3.fetch(any(S3EventNotificationRecord.class))) .thenThrow(new AmazonS3ConstraintException()); message.setBody(FixtureHelpers.fixture("fixtures/sns_notification.json")); final boolean actual = processor.test(message); verify(broadcaster, times(2)).isEmpty(); verify(broadcaster, never()).test(anyString()); verify(s3).fetch(any(S3EventNotificationRecord.class)); assertThat(actual).isTrue(); }
@Test public void testProcessSNS() throws Exception { final HttpRequestBase request = mock(HttpRequestBase.class); final S3ObjectInputStream stream = new S3ObjectInputStream( Resources.asByteSource(Resources.getResource("fixtures/s3_object.txt.gz")).openStream(), request); final ObjectMetadata metadata = new ObjectMetadata(); metadata.setContentEncoding("gzip"); final S3Object object = new S3Object(); object.setObjectMetadata(metadata); object.setObjectContent(stream); when(broadcaster.isEmpty()).thenReturn(false); when(s3.fetch(any(S3EventNotificationRecord.class))).thenReturn(object); message.setBody(FixtureHelpers.fixture("fixtures/sns_notification.json")); final boolean actual = processor.test(message); verify(broadcaster, times(2)).isEmpty(); verify(broadcaster, times(10)).test(anyString()); verify(s3).fetch(any(S3EventNotificationRecord.class)); verify(request, never()).abort(); assertThat(actual).isTrue(); }
@Test public void testProcessSQS() throws Exception { final HttpRequestBase request = mock(HttpRequestBase.class); final S3ObjectInputStream stream = new S3ObjectInputStream( Resources.asByteSource(Resources.getResource("fixtures/s3_object.txt.gz")).openStream(), request); final ObjectMetadata metadata = new ObjectMetadata(); metadata.setContentEncoding("gzip"); final S3Object object = new S3Object(); object.setObjectMetadata(metadata); object.setObjectContent(stream); when(broadcaster.isEmpty()).thenReturn(false); when(s3.fetch(any(S3EventNotificationRecord.class))).thenReturn(object); message.setBody(FixtureHelpers.fixture("fixtures/sqs_records.json")); final boolean actual = processor.test(message); verify(broadcaster, times(2)).isEmpty(); verify(broadcaster, times(10)).test(anyString()); verify(s3).fetch(any(S3EventNotificationRecord.class)); verify(request, never()).abort(); assertThat(actual).isTrue(); }
@Test public void testProcessNoConnectionsDuringDownload() throws Exception { final HttpRequestBase request = mock(HttpRequestBase.class); final S3ObjectInputStream stream = new S3ObjectInputStream( Resources.asByteSource(Resources.getResource("fixtures/s3_object.txt.gz")).openStream(), request); final ObjectMetadata metadata = new ObjectMetadata(); metadata.setContentEncoding("gzip"); final S3Object object = new S3Object(); object.setObjectMetadata(metadata); object.setObjectContent(stream); when(broadcaster.test(anyString())).thenReturn(false, false, false, false, true); when(s3.fetch(any(S3EventNotificationRecord.class))).thenReturn(object); message.setBody(FixtureHelpers.fixture("fixtures/sns_notification.json")); final boolean actual = processor.test(message); verify(broadcaster, times(2)).isEmpty(); verify(broadcaster, times(5)).test(anyString()); verify(s3).fetch(any(S3EventNotificationRecord.class)); verify(request).abort(); assertThat(actual).isFalse(); }