NakadiReader implements IORunnable { @Override public void run() throws IOException { try { runInternal(); } catch (BackoffException e) { throw e.getCause(); } } NakadiReader(URI uri, RequestFactory requestFactory, BackoffStrategy backoffStrategy, CursorManager cursorManager, ObjectMapper objectMapper, Set<String> eventNames, Optional<Subscription> subscription, Optional<Lock> lock, Class<T> eventClass, Listener<T> listener);   NakadiReader(URI uri, RequestFactory requestFactory, BackoffStrategy backoffStrategy, CursorManager cursorManager, Set<String> eventNames, Optional<Subscription> subscription, Optional<Lock> lock, EventReader<T> eventReader, Listener<T> listener, BatchHandler batchHandler, final MetricsCollector metricsCollector); @Override void run();  }
@Test public void shouldNotRetryInitialConnection() throws IOException { final Request request = mock(Request.class); when(request.execute()).thenThrow(new IOException("Initial connection failed")); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); expectedException.expect(IOException.class); expectedException.expectMessage(equalTo("Initial connection failed")); nakadiReader.run(); }
@Test public void shouldBeInterruptible() throws IOException, InterruptedException, BackoffException, ExecutionException, TimeoutException { final Response response = mock(Response.class); final InputStream endlessInputStream = new SequenceInputStream(new Enumeration<InputStream>() { @Override public boolean hasMoreElements() { return !Thread.currentThread().isInterrupted(); } @Override public InputStream nextElement() { try { return new InterruptibleInputStream(new ByteArrayInputStream("{\"cursor\":{\"partition\":\"0\",\"offset\":\"0\"}}".getBytes("utf-8"))); } catch (UnsupportedEncodingException e) { throw new UncheckedIOException(e); } } }); when(response.getBody()).thenReturn(endlessInputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2); final Future<?> future = scheduledExecutorService.submit(() -> { final Thread currentThread = Thread.currentThread(); scheduledExecutorService.schedule(currentThread::interrupt, 100, TimeUnit.MILLISECONDS); nakadiReader.unchecked().run(); }); Assert.assertNull("Thread should have completed normally", future.get()); }
@Test(timeout = 2000) public void shouldBeInterruptibleWhenReadingFromSocket() throws IOException, InterruptedException, BackoffException, ExecutionException, TimeoutException { final Response response = mock(Response.class); final InetAddress loopbackAddress = InetAddress.getLoopbackAddress(); final ServerSocket serverSocket = new ServerSocket(0, 0, loopbackAddress); final ExecutorService executorService = Executors.newSingleThreadExecutor(); executorService.submit(() -> { try { try (final Socket socket = serverSocket.accept()) { try (OutputStream out = socket.getOutputStream()) { while (true) { out.write("{\"cursor\":{\"partition\":\"0\",\"offset\":\"0\"}}\n".getBytes("utf-8")); try { Thread.sleep(100); } catch (InterruptedException e) { break; } } } } } catch (IOException e) { throw new UncheckedIOException(e); } }); final int localPort = serverSocket.getLocalPort(); final Socket socket = new Socket(loopbackAddress, localPort); socket.setSoTimeout(1000); final InputStream inputStream = socket.getInputStream(); when(response.getBody()).thenReturn(inputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final BackoffStrategy backoffStrategy = new NoBackoffStrategy(); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2); final Future<?> future = scheduledExecutorService.submit(() -> { final Thread currentThread = Thread.currentThread(); scheduledExecutorService.schedule(currentThread::interrupt, 100, TimeUnit.MILLISECONDS); nakadiReader.unchecked().run(); }); Assert.assertNull("Thread should have completed normally", future.get(500, TimeUnit.MILLISECONDS)); }
@Test public void shouldSendCursorsForLockedPartitions() throws IOException { final Request request = mock(Request.class); when(request.execute()).thenThrow(new IOException("Initial connection failed")); final Headers headers = new HeadersImpl(); when(request.getHeaders()).thenReturn(headers); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); when(cursorManager.getCursors(EVENT_NAME)).thenReturn(asList(new Cursor("0", "0"), new Cursor("1", "10"), new Cursor("2", "20"), new Cursor("3", "30"))); final Lock lock = new Lock(EVENT_NAME, "test", asList(new Partition("0", "0", "100"), new Partition("1", "0", "100"))); final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.of(lock), SomeEvent.class, listener); expectedException.expect(IOException.class); expectedException.expectMessage(equalTo("Initial connection failed")); nakadiReader.run(); assertEquals(singletonList("[{\"partition\":\"0\",\"offset\":\"0\"},{\"partition\":\"1\",\"offset\":\"10\"}"), headers.get("X-Nakadi-Cursors")); }
@Test public void shouldCloseOnIOException() throws IOException { final Response response = mock(Response.class); final byte[] bytes = "{\"cursor\":{\"partition\":\"123\",\"offset\":\"456\"},\"events\":[{\"id\":\"789\"}]}".getBytes("utf-8"); when(response.getBody()).thenReturn(new ByteArrayInputStream(bytes), new ByteArrayInputStream(bytes), new ByteArrayInputStream(bytes), new ByteArrayInputStream(bytes)); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); final Headers headers = new HeadersImpl(); when(request.getHeaders()).thenReturn(headers); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final ExponentialBackoffStrategy backoffStrategy = new EqualJitterBackoffStrategy().withMaxRetries(2); final Listener<SomeEvent> listener = events -> { throw new FileNotFoundException("from listener"); }; final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); try { nakadiReader.run(); } catch (FileNotFoundException e) { assertEquals("from listener", e.getMessage()); verify(request, times(3)).execute(); verify(response, times(3)).close(); } }
@Test public void shouldCloseOnRuntimeException() throws IOException { final Response response = mock(Response.class); final ByteArrayInputStream inputStream = new ByteArrayInputStream("{\"cursor\":{\"partition\":\"123\",\"offset\":\"456\"},\"foo\":\"bar\",\"events\":[{\"id\":\"789\"}],\"metadata\":{\"foo\":\"bar\"}}".getBytes("utf-8")); when(response.getBody()).thenReturn(inputStream); final Request request = mock(Request.class); when(request.execute()).thenReturn(response); final Headers headers = new HeadersImpl(); when(request.getHeaders()).thenReturn(headers); when(RequestFactory.createRequest(uri, "GET")).thenReturn(request); final NoBackoffStrategy backoffStrategy = new NoBackoffStrategy(); final Listener<SomeEvent> listener = events -> { throw new RuntimeException("from listener"); }; final NakadiReader<SomeEvent> nakadiReader = new NakadiReader<>(uri, RequestFactory, backoffStrategy, cursorManager, objectMapper, Collections.singleton(EVENT_NAME), Optional.empty(), Optional.empty(), SomeEvent.class, listener); try { nakadiReader.run(); } catch (RuntimeException e) { assertEquals("from listener", e.getMessage()); verify(request).execute(); verify(response).close(); } }