ProxyMonitorHandler extends AbstractProxyProtocolOptionHandler { @Override protected void doHandle(Channel channel, String[] content) { if(content == null || content.length < 1) { throw new IllegalArgumentException("monitor option is needed"); } ProxyMonitorParser.Type type = ProxyMonitorParser.Type.parse(content[0]); if(type == null) { logger.warn("[doHandle] unknown type: {}", content[0]); return; } switch (type) { case SocketStats: new SocketStatsResponser().response(channel); break; case PingStats: new PingStatsResponser().response(channel); break; case TunnelStats: new TunnelStatsResponser().response(channel); break; case TrafficStats: new TunnelTrafficResponser().response(channel); break; default: break; } } ProxyMonitorHandler(TunnelManager tunnelManager, PingStatsManager pingStatsManager, ProxyConfig proxyConfig); @Override PROXY_OPTION getOption();  }
@Test public void testPingStatsResponser() throws TimeoutException { PingStats pingStats = mock(PingStats.class); when(pingStats.getPingStatsResult()).thenReturn(new PingStatsResult(System.currentTimeMillis() - 2, System.currentTimeMillis(), localHostport(randomPort()), localHostport(randomPort()))); when(pingStatsManager.getAllPingStats()).thenReturn(Lists.newArrayList(pingStats)); AtomicReference<ByteBuf> result = new AtomicReference<>(); Channel channel = getWriteBackChannel(result); handler.doHandle(channel, new String[]{"PingStats"}); waitConditionUntilTimeOut(()->result.get() != null, 1000); ArrayParser parser = (ArrayParser) new ArrayParser().read(result.get()); Object[] objects = parser.getPayload(); for(Object object : objects) { PingStatsResult pingStatsResult = PingStatsResult.parse(object); logger.info("{}", pingStatsResult.getDirect()); logger.info("{}", pingStatsResult.getReal()); logger.info("{}", pingStatsResult.getStart()); logger.info("{}", pingStatsResult.getEnd()); } }
@Test public void testTunnelStatsResponser() throws Exception { TunnelStats tunnelStats = mock(TunnelStats.class); HostPort frontend = localHostport(randomPort()); HostPort backend = localHostport(randomPort()); Tunnel tunnel = tunnelManager.tunnels().get(0); TunnelStatsResult tunnelStatsResult = new TunnelStatsResult(tunnel.identity().toString(), new TunnelEstablished(null).name(), System.currentTimeMillis(), System.currentTimeMillis() + 10, frontend, backend); when(tunnelStats.getTunnelStatsResult()).thenReturn(tunnelStatsResult); when(tunnelMonitor.getTunnelStats()).thenReturn(tunnelStats); AtomicReference<ByteBuf> result = new AtomicReference<>(); Channel channel = getWriteBackChannel(result); handler.doHandle(channel, new String[]{"TunnelStats"}); waitConditionUntilTimeOut(()->result.get() != null, 1000); ArrayParser parser = (ArrayParser) new ArrayParser().read(result.get()); Object[] objects = parser.getPayload(); for(Object object : objects) { TunnelStatsResult other = TunnelStatsResult.parse(object); logger.info("{}", other.getTunnelId()); logger.info("{}", other.getTunnelState()); logger.info("{}", other.getProtocolRecvTime()); logger.info("{}", other.getProtocolSndTime()); logger.info("{}", other.getFrontend()); logger.info("{}", other.getBackend()); } }
@Test public void testTrafficStatsResponser() throws Exception { SessionTrafficResult frontend = new SessionTrafficResult(System.currentTimeMillis(), 100, 200, 10, 20); SessionTrafficResult backend = new SessionTrafficResult(System.currentTimeMillis(), 1000, 2000, 100, 200); Tunnel tunnel = tunnelManager.tunnels().get(0); TunnelTrafficResult trafficResult = new TunnelTrafficResult(tunnel.identity().toString(), frontend, backend); when(tunnel.getTunnelMonitor().getFrontendSessionMonitor().getSessionStats().getInputBytes()).thenReturn(frontend.getInputBytes()); when(tunnel.getTunnelMonitor().getFrontendSessionMonitor().getSessionStats().getOutputBytes()).thenReturn(frontend.getOutputBytes()); when(tunnel.getTunnelMonitor().getFrontendSessionMonitor().getSessionStats().lastUpdateTime()).thenReturn(frontend.getTimestamp()); when(tunnel.getTunnelMonitor().getBackendSessionMonitor().getSessionStats().getInputBytes()).thenReturn(backend.getInputBytes()); when(tunnel.getTunnelMonitor().getBackendSessionMonitor().getSessionStats().getOutputBytes()).thenReturn(backend.getOutputBytes()); when(tunnel.getTunnelMonitor().getBackendSessionMonitor().getSessionStats().lastUpdateTime()).thenReturn(backend.getTimestamp()); AtomicReference<ByteBuf> result = new AtomicReference<>(); Channel channel = getWriteBackChannel(result); handler.doHandle(channel, new String[]{"TrafficStats"}); waitConditionUntilTimeOut(()->result.get() != null, 1000); TunnelTrafficResult result1 = TunnelTrafficResult.parse(new ArrayParser().read(result.get()).getPayload()[0]); logger.info("{}", result1); }