HystrixConfigSseServlet extends HystrixSampleSseServlet { @Override protected int getNumberCurrentConnections() { return concurrentConnections.get(); } HystrixConfigSseServlet();   HystrixConfigSseServlet(Observable<HystrixConfiguration> sampleStream, int pausePollerThreadDelayInMs);   }
@Test public void testConfigDataWithStreamOnError() throws IOException, InterruptedException { servlet = new HystrixConfigSseServlet(streamOfOnNextThenOnError, 10); try { servlet.init(); } catch (ServletException ex) { } final AtomicInteger writes = new AtomicInteger(0); when(mockReq.getParameter("delay")).thenReturn("100"); when(mockResp.getWriter()).thenReturn(mockPrintWriter); Mockito.doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { String written = (String) invocation.getArguments()[0]; System.out.println("ARG : " + written); if (!written.contains("ping")) { writes.incrementAndGet(); } return null; } }).when(mockPrintWriter).print(Mockito.anyString()); Runnable simulateClient = new Runnable() { @Override public void run() { try { servlet.doGet(mockReq, mockResp); } catch (ServletException ex) { fail(ex.getMessage()); } catch (IOException ex) { fail(ex.getMessage()); } } }; Thread t = new Thread(simulateClient); t.start(); try { Thread.sleep(1000); System.out.println(System.currentTimeMillis() + " Woke up from sleep : " + Thread.currentThread().getName()); } catch (InterruptedException ex) { fail(ex.getMessage()); } assertEquals(1, writes.get()); assertEquals(0, servlet.getNumberCurrentConnections()); }
@Test public void testConfigDataWithStreamOnCompleted() throws IOException, InterruptedException { servlet = new HystrixConfigSseServlet(streamOfOnNextThenOnCompleted, 10); try { servlet.init(); } catch (ServletException ex) { } final AtomicInteger writes = new AtomicInteger(0); when(mockReq.getParameter("delay")).thenReturn("100"); when(mockResp.getWriter()).thenReturn(mockPrintWriter); Mockito.doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { String written = (String) invocation.getArguments()[0]; System.out.println("ARG : " + written); if (!written.contains("ping")) { writes.incrementAndGet(); } return null; } }).when(mockPrintWriter).print(Mockito.anyString()); Runnable simulateClient = new Runnable() { @Override public void run() { try { servlet.doGet(mockReq, mockResp); } catch (ServletException ex) { fail(ex.getMessage()); } catch (IOException ex) { fail(ex.getMessage()); } } }; Thread t = new Thread(simulateClient); t.start(); try { Thread.sleep(1000); System.out.println(System.currentTimeMillis() + " Woke up from sleep : " + Thread.currentThread().getName()); } catch (InterruptedException ex) { fail(ex.getMessage()); } assertEquals(1, writes.get()); assertEquals(0, servlet.getNumberCurrentConnections()); }
@Test public void testConfigDataWithIoExceptionOnWrite() throws IOException, InterruptedException { servlet = new HystrixConfigSseServlet(streamOfOnNexts, 10); try { servlet.init(); } catch (ServletException ex) { } final AtomicInteger writes = new AtomicInteger(0); when(mockResp.getWriter()).thenReturn(mockPrintWriter); Mockito.doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { String written = (String) invocation.getArguments()[0]; System.out.println("ARG : " + written); if (!written.contains("ping")) { writes.incrementAndGet(); } throw new IOException("simulated IO Exception"); } }).when(mockPrintWriter).print(Mockito.anyString()); Runnable simulateClient = new Runnable() { @Override public void run() { try { servlet.doGet(mockReq, mockResp); } catch (ServletException ex) { fail(ex.getMessage()); } catch (IOException ex) { fail(ex.getMessage()); } } }; Thread t = new Thread(simulateClient); t.start(); try { Thread.sleep(1000); System.out.println(System.currentTimeMillis() + " Woke up from sleep : " + Thread.currentThread().getName()); } catch (InterruptedException ex) { fail(ex.getMessage()); } assertTrue(writes.get() <= 2); assertEquals(0, servlet.getNumberCurrentConnections()); }
@Test public void testConfigDataWithInfiniteOnNextStream() throws IOException, InterruptedException { servlet = new HystrixConfigSseServlet(streamOfOnNexts, 10); try { servlet.init(); } catch (ServletException ex) { } final AtomicInteger writes = new AtomicInteger(0); when(mockReq.getParameter("delay")).thenReturn("100"); when(mockResp.getWriter()).thenReturn(mockPrintWriter); Mockito.doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { String written = (String) invocation.getArguments()[0]; System.out.println("ARG : " + written); if (!written.contains("ping")) { writes.incrementAndGet(); } return null; } }).when(mockPrintWriter).print(Mockito.anyString()); Runnable simulateClient = new Runnable() { @Override public void run() { try { servlet.doGet(mockReq, mockResp); } catch (ServletException ex) { fail(ex.getMessage()); } catch (IOException ex) { fail(ex.getMessage()); } } }; Thread t = new Thread(simulateClient); System.out.println("Starting thread : " + t.getName()); t.start(); System.out.println("Started thread : " + t.getName()); try { Thread.sleep(1000); System.out.println("Woke up from sleep : " + Thread.currentThread().getName()); } catch (InterruptedException ex) { fail(ex.getMessage()); } System.out.println("About to interrupt"); t.interrupt(); System.out.println("Done interrupting"); Thread.sleep(100); System.out.println("WRITES : " + writes.get()); assertTrue(writes.get() >= 9); assertEquals(0, servlet.getNumberCurrentConnections()); }