SNSS3Handler extends BaseHandler<SNSEvent> implements Handler<SNSEvent> { @Override public void handler(SNSEvent event, Context context) throws HandlerException { if (!initialized) { init(context); SNSS3HandlerConfig handlerConfig = (SNSS3HandlerConfig) this.config.getHandlerConfig(); this.logTrigger = handlerConfig.getLogSnsTrigger(); } this.source = this.sources.get(0); this.inputFiles = new ArrayList<String>(0); if (this.logTrigger) { logger.info("trigger: " + gson.toJson(event)); } for (SNSRecord record : event.getRecords()) { String json = record.getSNS().getMessage(); S3EventNotification s3Event = S3EventNotification.parseJson(json); List<S3EventNotificationRecord> toProcess = new ArrayList<S3EventNotificationRecord>(s3Event.getRecords()); for (S3EventNotificationRecord s3Record : s3Event.getRecords()) { String s3Path = String.format("s3: s3Record.getS3().getObject().getKey()); try { this.source = SourceUtils.getSource(s3Path, this.sources); } catch (SourceNotFoundException e) { logger.warn("skipping processing " + s3Path); toProcess.remove(s3Record); } } if (toProcess.size() == 0) { logger.warn("Nothing to process"); return; } this.inputFiles.addAll(toProcess.stream().map(m -> { return m.getS3().getObject().getKey(); }).collect(Collectors.toList())); this.recordIterator = new S3EventIterator(new LambdaContext(context), toProcess, s3ClientFactory); super.process(context); } }  @Override void handler(SNSEvent event, Context context); @Override Source getSource(); @Override String getSourceName(); @Override void onException(Exception e); @Override InternalEventIterator<InternalEvent> getInternalEventIterator();  }
@Test public void testExceptionHandlingd() throws Throwable { BaseHandler.CONFIG_FILE = "/com/nextdoor/bender/handler/config_test_sns.json"; TestContext ctx = new TestContext(); ctx.setFunctionName("unittest"); ctx.setInvokedFunctionArn("arn:aws:lambda:us-east-1:123:function:test-function:staging"); SNSS3Handler fhandler = (SNSS3Handler) getHandler(); fhandler.init(ctx); IpcSenderService ipcSpy = spy(fhandler.getIpcService()); doThrow(new TransportException("expected")).when(ipcSpy).flush(); fhandler.setIpcService(ipcSpy); AmazonSNSClient mockClient = mock(AmazonSNSClient.class); AmazonSNSClientFactory mockClientFactory = mock(AmazonSNSClientFactory.class); doReturn(mockClient).when(mockClientFactory).newInstance(); fhandler.snsClientFactory = mockClientFactory; SNSEvent event = getTestEvent(); try { fhandler.handler(event, ctx); } catch (Exception e) { } verify(mockClient, times(1)).publish("foo", "basic_input.log", "SNSS3Handler Failed"); }
@Test public void testSourceRegexFail() throws Throwable { BaseHandler.CONFIG_FILE = "/com/nextdoor/bender/handler/config_s3_source.json"; TestContext ctx = new TestContext(); ctx.setFunctionName("unittest"); ctx.setInvokedFunctionArn("arn:aws:lambda:us-east-1:123:function:test-function:staging"); SNSS3Handler handler = (SNSS3Handler) getHandler(); handler.init(ctx); handler.handler(getTestEvent("foo", false), ctx); assertEquals(0, DummyTransportHelper.BufferedTransporter.output.size()); }
@Test public void testSourceRegex() throws Throwable { BaseHandler.CONFIG_FILE = "/com/nextdoor/bender/handler/config_s3_source.json"; TestContext ctx = new TestContext(); ctx.setFunctionName("unittest"); ctx.setInvokedFunctionArn("arn:aws:lambda:us-east-1:123:function:test-function:staging"); SNSS3Handler handler = (SNSS3Handler) getHandler(); handler.init(ctx); handler.handler(getTestEvent(), ctx); assertEquals(1, DummyTransportHelper.BufferedTransporter.output.size()); }