@Test public void directLoadTest() throws Exception { StreamConnector sa = new StreamConnector(); StreamConnector sb = new StreamConnector(); Farcall fa = new Farcall(new BossConnector(sa.getInputStream(), sb.getOutputStream())); Farcall fb = new Farcall(new BossConnector(sb.getInputStream(), sa.getOutputStream())); int [] counts = new int[2]; fa.start(command -> { if (command.getName().equals("fast")) { counts[0]++; return "fast done"; } else if (command.getName().equals("slow")) { Thread.sleep(3); counts[1]++; return "slow done"; } return null; }); fb.start(); ExecutorService es = Executors.newWorkStealingPool(); ArrayList<Long> times = new ArrayList<>(); for( int rep=0; rep < 7; rep++ ) { ArrayList<Future<?>> futures = new ArrayList<>(); counts[0] = counts[1] = 0; long t = StopWatch.measure(() -> { CompletableFuture<?> cf = new CompletableFuture<>(); for (int r = 0; r < 40; r++) { futures.add(es.submit(() -> { for (int i = 0; i < 10; i++) assertEquals("fast done", fb.send("fast").waitSuccess()); return null; })); futures.add(es.submit(() -> { for (int i = 0; i < 2; i++) assertEquals("slow done", fb.send("slow").waitSuccess()); return null; })); } futures.forEach(f -> { try { f.get(); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } }); }); times.add(t); } long t1 = times.get(2); long t2 = times.get(times.size()-1); long mean = (t1 + t2)/2; assertThat((double) Math.abs(t2-t1) / ((double) mean), CoreMatchers.is(lessThan(0.16)) ); }
@Override public void send(Map<String, Object> data) throws IOException { if( closed.get() ) throw new IOException("connection closed"); bossOut.write(data); }
BossConnector extends BasicConnector implements Connector { @Override public void send(Map<String, Object> data) throws IOException { if( closed.get() ) throw new IOException("connection closed"); bossOut.write(data); } }
BossConnector extends BasicConnector implements Connector { @Override public void send(Map<String, Object> data) throws IOException { if( closed.get() ) throw new IOException("connection closed"); bossOut.write(data); } BossConnector(InputStream in, OutputStream out); }
BossConnector extends BasicConnector implements Connector { @Override public void send(Map<String, Object> data) throws IOException { if( closed.get() ) throw new IOException("connection closed"); bossOut.write(data); } BossConnector(InputStream in, OutputStream out); @Override void send(Map<String, Object> data); @Override Map<String, Object> receive(); }
BossConnector extends BasicConnector implements Connector { @Override public void send(Map<String, Object> data) throws IOException { if( closed.get() ) throw new IOException("connection closed"); bossOut.write(data); } BossConnector(InputStream in, OutputStream out); @Override void send(Map<String, Object> data); @Override Map<String, Object> receive();  }