class 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;
        }

        // since device ids from messages maybe different, so we use the InsertPlan not InsertTabletPlan.
        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);
        }
    }

    public  PublishHandler(IoTDBConfig config);
    protected  PublishHandler(IPlanExecutor executor, PayloadFormatter payloadFormat);

}