PublishHandler extends AbstractInterceptHandler { @Override public void onPublish(InterceptPublishMessage msg) { String clientId = msg.getClientID(); ByteBuf payload = msg.getPayload(); String topic = msg.getTopicName(); String username = msg.getUsername(); MqttQoS qos = msg.getQos(); LOG.debug("Receive publish message. clientId: {}, username: {}, qos: {}, topic: {}, payload: {}", clientId, username, qos, topic, payload); List<Message> events = payloadFormat.format(payload); if (events == null) { return; } for (Message event : events) { if (event == null) { continue; } InsertRowPlan plan = new InsertRowPlan(); plan.setTime(event.getTimestamp()); plan.setMeasurements(event.getMeasurements().toArray(new String[event.getMeasurements().size()])); plan.setValues(event.getValues().toArray(new Object[event.getValues().size()])); plan.setDataTypes(new TSDataType[event.getValues().size()]); plan.setNeedInferType(true); boolean status = false; try { plan.setDeviceId(new PartialPath(event.getDevice())); status = executeNonQuery(plan); } catch (QueryProcessException | StorageGroupNotSetException | StorageEngineException | IllegalPathException e ) { LOG.warn( "meet error when inserting device {}, measurements {}, at time {}, because ", event.getDevice(), event.getMeasurements(), event.getTimestamp(), e); } LOG.debug("event process result: {}", status); } } PublishHandler(IoTDBConfig config); protected  PublishHandler(IPlanExecutor executor, PayloadFormatter payloadFormat); @Override String getID(); @Override void onPublish(InterceptPublishMessage msg);  }
@Test public void onPublish() throws Exception { IPlanExecutor executor = mock(IPlanExecutor.class); PayloadFormatter payloadFormat = PayloadFormatManager.getPayloadFormat("json"); PublishHandler handler = new PublishHandler(executor, payloadFormat); String payload = "{\n" + "\"device\":\"root.sg.d1\",\n" + "\"timestamp\":1586076045524,\n" + "\"measurements\":[\"s1\"],\n" + "\"values\":[0.530635]\n" + "}"; ByteBuf buf = Unpooled.copiedBuffer(payload, StandardCharsets.UTF_8); MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader("root.sg.d1", 1); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE, false, 1); MqttPublishMessage publishMessage = new MqttPublishMessage(fixedHeader, variableHeader, buf); InterceptPublishMessage message = new InterceptPublishMessage(publishMessage, null, null); handler.onPublish(message); verify(executor).processNonQuery(any(InsertRowPlan.class)); }