erlmqtt_pub

#! /usr/bin/env escript
%% -*- erlang -*-
%%! -noshell -pa ./ebin -pa deps/getopt/ebin

-define(OPTS,
        [
         {help, undefined, "help", boolean,
          "Display usage information and exit"},

         {host, $h, "host", {string, "localhost"},
          "MQTT server hostname or IP address"},
         {port, $p, "port", {integer, 1883},
          "MQTT server port number"},
         {password, $P, "passwd", {string, "guest"},
          "Password for authentication"},
         {username, $u, "user", {string, "guest"},
          "Username for authentication"},
         {keep_alive, $k, "keepalive", {integer, 60},
          "Period for keepalive ping, in seconds"},

         {topic, $t, "topic", string,
          "The topic to which to publish"},
         {qos_level, $q, "qos", {integer, 0},
          "QoS level for the publication"},
         {retain, $r, "retain", {boolean, false},
          "Set the retain flag"},

         {file, $f, "file", string,
          "Read the message from the file path following"},
         {message, $m, "message", string,
          "A message to publish"},
         {lines, $l, "stdin-line", {boolean, false},
          "Read from stdin and publish a message per line"},
         {null, $n, "null-message", {boolean, false},
          "Publish an empty message"}
        ]).

main(Argv) ->
    {ok, {Opts, _Args}} = getopt:parse(?OPTS, Argv),
    case proplists:get_value(help, Opts) of
        true ->
            usage(),
            halt(0);
        _ ->
            ok
    end,
    [Host, Port, QosInt, Retain, Topic] =
        [proplists:get_value(K, Opts) ||
            K <- [host, port, qos_level, retain, topic]],
    if Topic == undefined ->
            usage_error("A topic must be supplied with -t");
       true -> ok
    end,
    Sources0 = [{K, proplists:get_value(K, Opts, false)} ||
                   K <- [file, message, lines, null]],
    Sources = lists:filter(fun ({_, S}) -> S =/= false end, Sources0),
    Msg = case Sources of
              [{file, File}] ->
                  {ok, M} = file:read_file(File), M;
              [{message, M}] ->
                  M;
              [{lines, true}] ->
                  lines;
              [{null, true}] ->
                  <<>>;
              _ ->
                  usage_error(
                    "Exactly one of -f, -m, -l or -n must be given")
          end,
    ConnectOpts = [O || O <- Opts, erlmqtt:is_connect_option(O)],
    {ok, C} = erlmqtt:open_clean({Host, Port}, ConnectOpts),

    QoS = erlmqtt:qos_symbol(QosInt),
    io:format(standard_error, "Publishing to ~s~n", [Topic]),
    case Msg of
        lines ->
            publish_lines(C, Topic, [QoS, {retain, Retain}]);
        _ ->
            erlmqtt:publish_sync(C, Topic, Msg, [QoS, {retain, Retain}])
    end,
    erlmqtt:close(C).

usage() ->
    ScriptPath = escript:script_name(),
    Script = filename:basename(ScriptPath),
    getopt:usage(?OPTS, Script).

usage_error(Msg) ->
    io:format(standard_error, "~s~n~n", [Msg]),
    usage(),
    halt(1).

publish_lines(Conn, Topic, Opts) ->
    case io:get_line("") of
        eof ->
            ok;
        Msg0 ->
            %% ditch the line ending (or eof)
            Msg = lists:sublist(Msg0, length(Msg0) - 1),
            erlmqtt:publish_sync(Conn, Topic, Msg, Opts),
            publish_lines(Conn, Topic, Opts)
    end.