class NakadiReader implements IORunnable {

    @Override
    public void run() throws IOException {
        try {
            runInternal();
        } catch (BackoffException e) {
            throw e.getCause();
        }
    }

}

class NakadiReaderTest {

    private static final String EVENT_NAME;
    private final URI uri;
    private final ObjectMapper objectMapper;
    private final CursorManager cursorManager;
    private final RequestFactory RequestFactory;
    @SuppressWarnings("unchecked")
    private final Listener<SomeEvent> listener;
    @Rule
    public final ExpectedException expectedException;

    @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));
    }

}