PushAgent extends AbstractAgent { protected void startWriteHttpJsonListener(String strPort, ReportableEntityHandlerFactory handlerFactory) { final int port = Integer.parseInt(strPort); registerPrefixFilter(strPort); registerTimestampFilter(strPort); if (proxyConfig.isHttpHealthCheckAllPorts()) healthCheckManager.enableHealthcheck(port); ChannelHandler channelHandler = new WriteHttpJsonPortUnificationHandler(strPort, tokenAuthenticator, healthCheckManager, handlerFactory, proxyConfig.getHostname(), preprocessors.get(strPort)); startAsManagedThread(port, new TcpIngester(createInitializer(channelHandler, port, proxyConfig.getPushListenerMaxReceivedLength(), proxyConfig.getPushListenerHttpBufferSize(), proxyConfig.getListenerIdleConnectionTimeout(), getSslContext(strPort), getCorsConfig(strPort)), port). withChildChannelOptions(childChannelOptions), "listener-plaintext-writehttpjson-" + port); logger.info("listening on port: " + strPort + " for write_http data"); }  static void main(String[] args); @Override void stopListeners();  }
@Test public void testWriteHttpJsonMetricsPortHandler() throws Exception { port = findAvailablePort(4878); proxy.proxyConfig.writeHttpJsonListenerPorts = String.valueOf(port); proxy.proxyConfig.hostname = "defaultLocalHost"; proxy.startWriteHttpJsonListener(proxy.proxyConfig.writeHttpJsonListenerPorts, mockHandlerFactory); waitUntilListenerIsOnline(port); reset(mockPointHandler); mockPointHandler.reject((ReportPoint) eq(null), anyString()); expectLastCall().times(2); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("disk.sda.disk_octets.read").setHost("testSource"). setTimestamp(startTime * 1000).setValue(197141504).build()); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("disk.sda.disk_octets.write").setHost("testSource"). setTimestamp(startTime * 1000).setValue(175136768).build()); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("disk.sda.disk_octets.read").setHost("defaultLocalHost"). setTimestamp(startTime * 1000).setValue(297141504).build()); mockPointHandler.report(ReportPoint.newBuilder().setTable("dummy"). setMetric("disk.sda.disk_octets.write").setHost("defaultLocalHost"). setTimestamp(startTime * 1000).setValue(275136768).build()); replay(mockPointHandler); String payloadStr = "[{\n" + "\"values\": [197141504, 175136768],\n" + "\"dstypes\": [\"counter\", \"counter\"],\n" + "\"dsnames\": [\"read\", \"write\"],\n" + "\"time\": " + startTime + ",\n" + "\"interval\": 10,\n" + "\"host\": \"testSource\",\n" + "\"plugin\": \"disk\",\n" + "\"plugin_instance\": \"sda\",\n" + "\"type\": \"disk_octets\",\n" + "\"type_instance\": \"\"\n" + "},{\n" + "\"values\": [297141504, 275136768],\n" + "\"dstypes\": [\"counter\", \"counter\"],\n" + "\"dsnames\": [\"read\", \"write\"],\n" + "\"time\": " + startTime + ",\n" + "\"interval\": 10,\n" + "\"plugin\": \"disk\",\n" + "\"plugin_instance\": \"sda\",\n" + "\"type\": \"disk_octets\",\n" + "\"type_instance\": \"\"\n" + "},{\n" + "\"dstypes\": [\"counter\", \"counter\"],\n" + "\"dsnames\": [\"read\", \"write\"],\n" + "\"time\": " + startTime + ",\n" + "\"interval\": 10,\n" + "\"plugin\": \"disk\",\n" + "\"plugin_instance\": \"sda\",\n" + "\"type\": \"disk_octets\",\n" + "\"type_instance\": \"\"\n" + "}]\n"; assertEquals(400, gzippedHttpPost("http: assertEquals(200, gzippedHttpPost("http: verify(mockPointHandler); }