My MQTT protocol and EMQTTD open source project personal understanding (12) - Hook use and connect Kafka to send messages, using the EKAF library

First, work environment preparation

The Erlang Kafka client library uses github.com/helpshift/e…

Emq uses v2.3.5, github.com/emqtt/emq-r…

Kafka’s running environment, blog.csdn.net/libaineu200…

We are in the form of a plug-in, my plug-in path is/home/firecat/Prj/emq2.0 / emq – relx – 2.3.5 / deps

Copy a copy of emq_plugin_template and rename it to emq_plugin_kafka_ekaf. Note that the relevant configuration files and source files should be renamed.

Enter the directory/home/firecat/Prj/emq2.0 / emq – relx – 2.3.5 / deps/emq_plugin_kafka_ekaf \

Makefile file with new red text

BUILD_DEPS = emqttd cuttlefish ekaf dep_ekaf = git github.com/helpshift/e… master\

Ii. Modification of project documents

1, enter the directory/home/firecat/Prj/emq2.0 / emq – relx – 2.3.5 / Makefile \

The Makefile to increase

DEPS += emq_plugin_kafka_ekaf

Add three lines to relx.config release:

kafkamocker,

ekaf,

{emq_plugin_kafka_ekaf, load}\

3, / home/firecat/Prj/emq2.0 / emq relx – 2.3.5 / data/loaded_plugins set to start the plugin

emq_recon.

emq_modules.

emq_retainer.

emq_dashboard.

emq_plugin_kafka_ekaf.\

Three, source code implementation

1, / home/firecat/Prj/emq2.0 / emq relx – 2.3.5 deps/emq_plugin_kafka_ekaf/etc/emq_plugin_kafka_ekaf config

[{emq_plugin_kafka_ekaf, [{kafka, [{bootstrap_broker, {"127.0.0.1", 9092}}, %% cannot use "localhost" {partition_strategy, strict_round_robin} ]} ]} ].Copy the code

2, / home/firecat/Prj/emq2.0 / emq – relx – 2.3.5 / deps/emq_plugin_kafka_ekaf/SRC/emq_plugin_kafka_ekaf erl

%%--------------------------------------------------------------------
%% Copyright (c) 2013-2018 EMQ Enterprise, Inc. (http://emqtt.io)
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------

-module(emq_plugin_kafka_ekaf).

-include_lib("emqttd/include/emqttd.hrl").

-define(TEST_TOPIC,<<"emq_broker_message">>).

-export([load/1, unload/0]).

%% Hooks functions

-export([on_client_connected/3, on_client_disconnected/3]).

-export([on_client_subscribe/4, on_client_unsubscribe/4]).

-export([on_session_created/3, on_session_subscribed/4, on_session_unsubscribed/4, on_session_terminated/4]).

-export([on_message_publish/2, on_message_delivered/4, on_message_acked/4]).

%% Called when the plugin application start
load(Env) ->
    ekaf_init([Env]),
    emqttd:hook('client.connected', fun ?MODULE:on_client_connected/3, [Env]),
    emqttd:hook('client.disconnected', fun ?MODULE:on_client_disconnected/3, [Env]),
    emqttd:hook('client.subscribe', fun ?MODULE:on_client_subscribe/4, [Env]),
    emqttd:hook('client.unsubscribe', fun ?MODULE:on_client_unsubscribe/4, [Env]),
    emqttd:hook('session.created', fun ?MODULE:on_session_created/3, [Env]),
    emqttd:hook('session.subscribed', fun ?MODULE:on_session_subscribed/4, [Env]),
    emqttd:hook('session.unsubscribed', fun ?MODULE:on_session_unsubscribed/4, [Env]),
    emqttd:hook('session.terminated', fun ?MODULE:on_session_terminated/4, [Env]),
    emqttd:hook('message.publish', fun ?MODULE:on_message_publish/2, [Env]),
    emqttd:hook('message.delivered', fun ?MODULE:on_message_delivered/4, [Env]),
    emqttd:hook('message.acked', fun ?MODULE:on_message_acked/4, [Env]).

on_client_connected(ConnAck, Client = #mqtt_client{client_id = ClientId}, _Env) ->
    io:format("client ~s connected, connack: ~w~n", [ClientId, ConnAck]),
    Json = mochijson2:encode([
        {type, <<"connected">>},
        {client_id, ClientId},
        {cluster_node, node()},
        {ts, emqttd_time:now_ms()}
    ]),
    
    %%ekaf:produce_async_batched(?TEST_TOPIC, list_to_binary(Json)),批量发送，不建议使用，我发现kafka接收会滞后
    ekaf:produce_async(?TEST_TOPIC, list_to_binary(Json)),
    ekaf:produce_async(?TEST_TOPIC, {<<"mykey_1">>, list_to_binary(Json)}),
    {ok, Client}.

on_client_disconnected(Reason, _Client = #mqtt_client{client_id = ClientId}, _Env) ->
    io:format("client ~s disconnected, reason: ~w~n", [ClientId, Reason]),
    Json = mochijson2:encode([
        {type, <<"disconnected">>},
        {client_id, ClientId},
        {reason, Reason},
        {cluster_node, node()},
        {ts, emqttd_time:now_ms()}
    ]),

    %%ekaf:produce_async_batched(?TEST_TOPIC, list_to_binary(Json)),批量发送，不建议使用，我发现kafka接收会滞后
    ekaf:produce_async(?TEST_TOPIC, list_to_binary(Json)),
    ekaf:produce_async(?TEST_TOPIC, {<<"mykey_2">>, list_to_binary(Json)}),
    ok.

on_client_subscribe(ClientId, Username, TopicTable, _Env) ->
    io:format("client(~s/~s) will subscribe: ~p~n", [Username, ClientId, TopicTable]),
    {ok, TopicTable}.
    
on_client_unsubscribe(ClientId, Username, TopicTable, _Env) ->
    io:format("client(~s/~s) unsubscribe ~p~n", [ClientId, Username, TopicTable]),
    {ok, TopicTable}.

on_session_created(ClientId, Username, _Env) ->
    io:format("session(~s/~s) created.", [ClientId, Username]).

on_session_subscribed(ClientId, Username, {Topic, Opts}, _Env) ->
    io:format("session(~s/~s) subscribed: ~p~n", [Username, ClientId, {Topic, Opts}]),
    {ok, {Topic, Opts}}.

on_session_unsubscribed(ClientId, Username, {Topic, Opts}, _Env) ->
    io:format("session(~s/~s) unsubscribed: ~p~n", [Username, ClientId, {Topic, Opts}]),
    ok.

on_session_terminated(ClientId, Username, Reason, _Env) ->
    io:format("session(~s/~s) terminated: ~p.", [ClientId, Username, Reason]).

%% transform message and return
on_message_publish(Message = #mqtt_message{topic = <<"$SYS/", _/binary>>}, _Env) ->
    {ok, Message};

on_message_publish(Message, _Env) ->
    io:format("publish ~s~n", [emqttd_message:format(Message)]),
    
    Id = Message#mqtt_message.id,
    From = Message#mqtt_message.from, %需要登录和不需要登录这里的返回值是不一样的
    Topic = Message#mqtt_message.topic,
    Payload = Message#mqtt_message.payload,
    Qos = Message#mqtt_message.qos,
    Dup = Message#mqtt_message.dup,
    Retain = Message#mqtt_message.retain,
    Timestamp = Message#mqtt_message.timestamp,

    ClientId = c(From),
    Username = u(From),

    Json = mochijson2:encode([
			      {type, <<"publish">>},
			      {client_id, ClientId},
			      {message, [
					 {username, Username},
					 {topic, Topic},
					 {payload, Payload},
					 {qos, i(Qos)},
					 {dup, i(Dup)},
					 {retain, i(Retain)}
					]},
			      {cluster_node, node()},
			      {ts, emqttd_time:now_ms()}
			     ]),

    %%ekaf:produce_async_batched(?TEST_TOPIC, list_to_binary(Json)),批量发送，不建议使用，我发现kafka接收会滞后
    ekaf:produce_async(?TEST_TOPIC, list_to_binary(Json)),

    {ok, Message}.

on_message_delivered(ClientId, Username, Message, _Env) ->
    io:format("delivered to client(~s/~s): ~s~n", [Username, ClientId, emqttd_message:format(Message)]),
    {ok, Message}.

on_message_acked(ClientId, Username, Message, _Env) ->
    io:format("client(~s/~s) acked: ~s~n", [Username, ClientId, emqttd_message:format(Message)]),
    {ok, Message}.

%% Called when the plugin application stop
unload() ->
    emqttd:unhook('client.connected', fun ?MODULE:on_client_connected/3),
    emqttd:unhook('client.disconnected', fun ?MODULE:on_client_disconnected/3),
    emqttd:unhook('client.subscribe', fun ?MODULE:on_client_subscribe/4),
    emqttd:unhook('client.unsubscribe', fun ?MODULE:on_client_unsubscribe/4),
    emqttd:unhook('session.created', fun ?MODULE:on_session_created/3),
    emqttd:unhook('session.subscribed', fun ?MODULE:on_session_subscribed/4),
    emqttd:unhook('session.unsubscribed', fun ?MODULE:on_session_unsubscribed/4),
    emqttd:unhook('session.terminated', fun ?MODULE:on_session_terminated/4),
    emqttd:unhook('message.publish', fun ?MODULE:on_message_publish/2),
    emqttd:unhook('message.delivered', fun ?MODULE:on_message_delivered/4),
    emqttd:unhook('message.acked', fun ?MODULE:on_message_acked/4).

%% ===================================================================
%% ekaf_init https://github.com/helpshift/ekaf
%% ===================================================================
ekaf_init(_Env) ->
    %% Start
    application:load(ekaf),
    %% Get parameters
    {ok, Kafka} = application:get_env(?MODULE, kafka),
    BootstrapBroker = proplists:get_value(bootstrap_broker, Kafka),
    PartitionStrategy= proplists:get_value(partition_strategy, Kafka),
    %% Set partition strategy, like application:set_env(ekaf, ekaf_partition_strategy, strict_round_robin),
    application:set_env(ekaf, ekaf_partition_strategy, PartitionStrategy),
    %% Set broker url and port, like application:set_env(ekaf, ekaf_bootstrap_broker, {"127.0.0.1", 9092}),
    application:set_env(ekaf, ekaf_bootstrap_broker, BootstrapBroker),
    %% Set topic
    application:set_env(ekaf, ekaf_bootstrap_topics, ?TEST_TOPIC),

    {ok, _} = application:ensure_all_started(kafkamocker),
    {ok, _} = application:ensure_all_started(gproc),
    {ok, _} = application:ensure_all_started(ekaf),

    io:format("Init ekaf with ~p~n", [BootstrapBroker]).

i(true) -> 1;
i(false) -> 0;
i(I) when is_integer(I) -> I.
c({ClientId, Username}) -> ClientId;
c(From) -> From.
u({ClientId, Username}) -> Username;
u(From) -> From.
Copy the code

Complete source code download address: download.csdn.net/download/li…

—

Refer to the article: www.cnblogs.com/wunaozai/p/…

Github.com/msdevanms/e… \

mo4tech.com (Moment For Technology) is a global community with thousands techies from across the global hang out!Passionate technologists, be it gadget freaks, tech enthusiasts, coders, technopreneurs, or CIOs, you would find them all here.

My MQTT protocol and EMQTTD open source project personal understanding (12) – Hook use and connect Kafka to send messages, using the EKAF library

My MQTT protocol and EMQTTD open source project personal understanding (12) – Hook use and connect Kafka to send messages, using the EKAF library

Related Posts

Autumn recruit will algorithm problem – prefix and

Fourier Transform: Python signal processing using scipy. FFT

MySQL | | error: SQL_ERROR_INFO: ‘Every derived table must have its own alias’