diff --git a/ChangeLog.md b/ChangeLog.md new file mode 100644 --- /dev/null +++ b/ChangeLog.md @@ -0,0 +1,3 @@ +# Changelog for mqtt-rpc + +## Unreleased changes diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright Dustin Sallings (c) 2019 + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Dustin Sallings nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,15 @@ +# mqtt-rpc + +This package provides an RPC client interface over MQTT. + +It's currently quite low-level and not very configurable, but I'm +using it in a real application, so it at least needs to work for me. +:) + +## Example: + +```haskell + mc <- connectURI mqttConfig{_protocol=Protocol50} "mqtt://broker/" + response <- call mc "some/path" "a message" + BL.hPut stdout response +``` diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/app/Main.hs b/app/Main.hs new file mode 100644 --- /dev/null +++ b/app/Main.hs @@ -0,0 +1,38 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Main where + +import qualified Data.ByteString.Lazy as BL +import Data.Maybe (fromJust) +import Network.MQTT.Client +import Network.MQTT.RPC +import Network.URI +import Options.Applicative (Parser, execParser, fullDesc, help, + helper, info, long, maybeReader, option, + progDesc, short, showDefault, strOption, + value, (<**>)) +import System.IO (stdout) + +data Options = Options { + optUri :: URI + , optTopic :: Topic + , optMsg :: BL.ByteString + } + +options :: Parser Options +options = Options + <$> option (maybeReader parseURI) (long "mqtt-uri" <> short 'u' <> showDefault <> value (fromJust $ parseURI "mqtt://localhost/") <> help "mqtt broker URI") + <*> strOption (long "mqtt-topic" <> short 't' <> showDefault <> value "tmp/rpctest" <> help "MQTT request topic") + <*> strOption (long "mqtt-msg" <> short 'm' <> showDefault <> value "" <> help "Request Message") + +run :: Options -> IO () +run Options{..} = do + mc <- connectURI mqttConfig{_protocol=Protocol50} optUri + BL.hPut stdout =<< call mc optTopic optMsg + +main :: IO () +main = run =<< execParser opts + + where opts = info (options <**> helper) (fullDesc <> progDesc "RPC test") diff --git a/net-mqtt-rpc.cabal b/net-mqtt-rpc.cabal new file mode 100644 --- /dev/null +++ b/net-mqtt-rpc.cabal @@ -0,0 +1,68 @@ +cabal-version: 1.12 + +-- This file has been generated from package.yaml by hpack version 0.31.2. +-- +-- see: https://github.com/sol/hpack +-- +-- hash: 2c985a6c1130dcb54d621da343ccbfb4aefad30a6fab33e6bd43c33f66915674 + +name: net-mqtt-rpc +version: 0.1.0.0 +synopsis: Make RPC calls via an MQTT broker. +description: Please see the README on GitHub at +category: Network +homepage: https://github.com/dustin/net-mqtt-rpc#readme +bug-reports: https://github.com/dustin/net-mqtt-rpc/issues +author: Dustin Sallings +maintainer: dustin@spy.net +copyright: MIT +license: BSD3 +license-file: LICENSE +build-type: Simple +extra-source-files: + README.md + ChangeLog.md + +source-repository head + type: git + location: https://github.com/dustin/net-mqtt-rpc + +library + exposed-modules: + Network.MQTT.RPC + other-modules: + Paths_net_mqtt_rpc + hs-source-dirs: + src + default-extensions: OverloadedStrings RecordWildCards NamedFieldPuns + ghc-options: -Wall + build-depends: + base >=4.7 && <5 + , bytestring + , net-mqtt >=0.6.2.0 && <0.7.0.0 + , random + , stm + , text + , uuid + default-language: Haskell2010 + +executable mqtt-rpc + main-is: Main.hs + other-modules: + Paths_net_mqtt_rpc + hs-source-dirs: + app + default-extensions: OverloadedStrings RecordWildCards NamedFieldPuns + ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall + build-depends: + base >=4.7 && <5 + , bytestring + , net-mqtt >=0.6.2.0 && <0.7.0.0 + , net-mqtt-rpc + , network-uri + , optparse-applicative + , random + , stm + , text + , uuid + default-language: Haskell2010 diff --git a/src/Network/MQTT/RPC.hs b/src/Network/MQTT/RPC.hs new file mode 100644 --- /dev/null +++ b/src/Network/MQTT/RPC.hs @@ -0,0 +1,52 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} + +module Network.MQTT.RPC (call) where + +import Control.Concurrent.STM (atomically, newTChanIO, + readTChan, writeTChan) +import qualified Control.Exception as E +import qualified Data.ByteString.Lazy as BL +import Control.Monad (when) +import Data.Text (Text) +import qualified Data.Text.Encoding as TE +import qualified Data.UUID as UUID +import Network.MQTT.Client +import System.Random (randomIO) + +blToText :: BL.ByteString -> Text +blToText = TE.decodeUtf8 . BL.toStrict + +-- | Send a message to a topic on an MQTT broker with a random +-- subscription and correlation such that an agent may receive this +-- message and respond over the ephemeral channel. The response will +-- be returned. +-- +-- Note that this client provides no timeouts or retries. MQTT will +-- guarantee the request message is delivered to the broker, but if +-- there's nothing to pick it up, there may never be a response. +call :: MQTTClient -> Topic -> BL.ByteString -> IO BL.ByteString +call mc topic req = do + r <- newTChanIO + corr <- BL.fromStrict . UUID.toASCIIBytes <$> randomIO + subid <- BL.fromStrict . ("$rpc/" <>) . UUID.toASCIIBytes <$> randomIO + go corr subid r + + where go theID theTopic r = E.bracket reg unreg rt + where reg = do + atomically $ registerCorrelated mc theID (SimpleCallback cb) + subscribe mc [(blToText theTopic, subOptions)] mempty + unreg _ = do + atomically $ unregisterCorrelated mc theID + unsubscribe mc [blToText theTopic] mempty + cb _ _ m _ = atomically $ writeTChan r m + rt _ = do + publishq mc topic req False QoS2 [ + PropCorrelationData theID, + PropResponseTopic theTopic] + atomically do + connd <- isConnectedSTM mc + when (not connd) $ E.throw (MQTTException "disconnected") + readTChan r