diff --git a/hcoap.cabal b/hcoap.cabal --- a/hcoap.cabal +++ b/hcoap.cabal @@ -1,6 +1,6 @@ copyright: Copyright (c) 2016, Ulf Lilleengen name: hcoap -version: 0.1.0.2 +version: 0.1.1.0 synopsis: Library for handling CoAP protocol license: BSD3 license-file: LICENSE @@ -45,6 +45,7 @@ QuickCheck, HUnit, random, + async, bytestring ghc-options: -threaded -rtsopts -with-rtsopts=-N default-language: Haskell2010 diff --git a/src/Network/CoAP/Client.hs b/src/Network/CoAP/Client.hs --- a/src/Network/CoAP/Client.hs +++ b/src/Network/CoAP/Client.hs @@ -14,16 +14,14 @@ -} module Network.CoAP.Client ( Request(..) -, Client +, Client(..) , Method(..) , Response(..) , ResponseCode(..) , Option(..) , OptionString , MediaType(..) -, doRequest , createClient -, shutdownClient ) where import Network.CoAP.Messaging @@ -39,20 +37,24 @@ -- | A client that can perform CoAP requests. data Client = Client { -- | Send a CoAP request to a given endpoint. Returns a CoAP response. doRequest :: Endpoint -> Request -> IO Response - , msgThreadId :: Async ()} + -- | Stop a client. Ensures that no threads are running and that messaging + -- layer is shut down. + , shutdownClient :: IO () } -- | Create a client using a given transport. This will spawn internal messaging threads making the -- client ready to send requests. createClient :: Transport -> IO Client createClient transport = do state <- createMessagingState transport - msgThread <- async (messagingLoop state) + msgThreads <- startMessaging state return Client { doRequest = doRequestInternal state - , msgThreadId = msgThread } + , shutdownClient = stopClient state msgThreads } -- | Shuts down the internal messaging threads and stops the client -shutdownClient :: Client -> IO () -shutdownClient client = wait (msgThreadId client) -- TODO: Actually shut down the threads... +stopClient :: MessagingState -> [Async ()] -> IO () +stopClient state threads = do + stopMessaging state threads + mapM_ wait threads generateToken :: Int -> IO [Word8] generateToken 0 = return [] @@ -63,19 +65,19 @@ doRequestInternal :: MessagingState -> Endpoint -> Request -> IO Response doRequestInternal state dest (Request method options payload reliable) = do - let header = MessageHeader { messageVersion = 1 - , messageType = if reliable then CON else NON - , messageCode = CodeRequest method - , messageId = 0 } + tokenLen <- randomRIO (0, 8) token <- generateToken tokenLen - let msg = Message { messageHeader = header - , messageToken = pack token - , messageOptions = options - , messagePayload = payload } + let msg = Message { messageVersion = 1 + , messageType = if reliable then CON else NON + , messageCode = CodeRequest method + , messageId = 0 + , messageToken = pack token + , messageOptions = options + , messagePayload = payload } sendRequest msg dest state responseCtx <- recvResponse msg dest state - let (Message (MessageHeader _ _ (CodeResponse rCode) _) _ opts pload) = message responseCtx + let (Message _ _ (CodeResponse rCode) _ _ opts pload) = message responseCtx return Response { responseCode = rCode , responseOptions = opts , responsePayload = pload } diff --git a/src/Network/CoAP/Message.hs b/src/Network/CoAP/Message.hs --- a/src/Network/CoAP/Message.hs +++ b/src/Network/CoAP/Message.hs @@ -8,7 +8,6 @@ -} module Network.CoAP.Message ( Message(..) -, MessageHeader(..) , MessageVersion(..) , MessageType(..) , MessageId(..) diff --git a/src/Network/CoAP/MessageCodec.hs b/src/Network/CoAP/MessageCodec.hs --- a/src/Network/CoAP/MessageCodec.hs +++ b/src/Network/CoAP/MessageCodec.hs @@ -18,24 +18,24 @@ import Prelude hiding (null, length, fromStrict, toStrict) getType :: Word8 -> Get MessageType -getType 0 = return (CON) -getType 1 = return (NON) -getType 2 = return (ACK) -getType 3 = return (RST) +getType 0 = return CON +getType 1 = return NON +getType 2 = return ACK +getType 3 = return RST getType _ = fail "Unknown type" getRequestMethod :: Word8 -> Get Method -getRequestMethod 1 = return (GET) -getRequestMethod 2 = return (POST) -getRequestMethod 3 = return (PUT) -getRequestMethod 4 = return (DELETE) +getRequestMethod 1 = return GET +getRequestMethod 2 = return POST +getRequestMethod 3 = return PUT +getRequestMethod 4 = return DELETE getRequestMethod _ = fail "Unknown request method" getResponseCode :: Word8 -> Get ResponseCode -getResponseCode _ = return (Created) +getResponseCode _ = return Created getCode :: Word8 -> Word8 -> Get MessageCode -getCode 0 0 = return (CodeEmpty) +getCode 0 0 = return CodeEmpty getCode 0 detail = do method <- getRequestMethod detail return (CodeRequest method) @@ -52,28 +52,11 @@ getTokenLength len = if len > 8 then fail "Invalid token length" - else return (len) - -getHeader :: Get (MessageHeader, Word8) -getHeader = do - tmp <- getWord8 - let version = fromIntegral (shiftR ((.&.) tmp 0xC0) 6) - msgType <- getType (shiftR ((.&.) tmp 0x30) 4) - tokenLength <- getTokenLength ((.&.) tmp 0x0F) - - c <- getWord8 - let clazz = fromIntegral (shiftR ((.&.) c 0x70) 5) - let detail = fromIntegral ((.&.) c 0x1F) - code <- getCode clazz detail - - id <- getWord16be - return ((MessageHeader version msgType code (fromIntegral id)), tokenLength) + else return len -getToken :: Word8 -> Get (Token) +getToken :: Word8 -> Get Token getToken 0 = return BS.empty -getToken n = do - str <- getByteString (fromIntegral n) - return str +getToken n = getByteString (fromIntegral n) intToMediaType 0 = TextPlain intToMediaType 40 = ApplicationLinkFormat @@ -100,41 +83,24 @@ decodeOption 60 value valueLen = Size1 (decodeOptionInt value valueLen) getOptionInt :: Int -> Get Int -getOptionInt valueLen = do - if valueLen == 0 - then return 0 - else if valueLen < 256 - then do - v <- getWord8 - return (fromIntegral v) - else if valueLen < 65536 - then do - v <- getWord16be - return (fromIntegral v) - else if valueLen < 16777216 - then do - a <- getWord8 - b <- getWord8 - c <- getWord8 - return ((.|.) ((.|.) (shiftL (fromIntegral a :: Int) 16) (shiftL (fromIntegral b :: Int) 8)) (fromIntegral c :: Int)) - else do - v <- getWord32be - return (fromIntegral v) +getOptionInt valueLen + | valueLen == 0 = return 0 + | valueLen < 256 = do v <- getWord8; return (fromIntegral v) + | valueLen < 65536 = do v <- getWord16be; return (fromIntegral v) + | valueLen < 16777216 = do a <- getWord8 + b <- getWord8 + c <- getWord8 + return ((.|.) ((.|.) (shiftL (fromIntegral a :: Int) 16) (shiftL (fromIntegral b :: Int) 8)) (fromIntegral c :: Int)) + | otherwise = do v <- getWord32be; return (fromIntegral v) decodeOptionInt :: BS.ByteString -> Int -> Int decodeOptionInt value valueLen = runGet (getOptionInt valueLen) (fromStrict value) -getOptionPart :: Int -> Get (Int) -getOptionPart part = do - if part == 14 - then do - v <- getWord16be - return ((fromIntegral v) + 269) - else if part == 13 - then do - v <- getWord8 - return ((fromIntegral v) + 13) - else return part +getOptionPart :: Int -> Get Int +getOptionPart part + | part == 14 = do v <- getWord16be; return (fromIntegral v + 269) + | part == 13 = do v <- getWord8; return (fromIntegral v + 13) + | otherwise = return part getOption :: Int -> Get (Maybe Option) getOption lastCode = do @@ -155,18 +121,17 @@ let option = decodeOption optCode value fullLength return (Just option) -getOptionsLoop :: Int -> Get ([Option]) +getOptionsLoop :: Int -> Get [Option] getOptionsLoop lastCode = do opt <- getOption lastCode case opt of - Nothing -> do - return [] + Nothing -> return [] Just o -> do let optCode = fst (encodeOption o) opts <- getOptionsLoop optCode return (o:opts) -getOptions :: Get ([Option]) +getOptions :: Get [Option] getOptions = getOptionsLoop 0 getPayload :: Get (Maybe Payload) @@ -176,17 +141,28 @@ then Nothing else Just (BS.pack (unpack str))) - getMessage :: Get Message getMessage = do - (header, tokenLength) <- getHeader + tmp <- getWord8 + let version = fromIntegral (shiftR ((.&.) tmp 0xC0) 6) + msgType <- getType (shiftR ((.&.) tmp 0x30) 4) + tokenLength <- getTokenLength ((.&.) tmp 0x0F) + + c <- getWord8 + let clazz = fromIntegral (shiftR ((.&.) c 0x70) 5) + let detail = fromIntegral ((.&.) c 0x1F) + code <- getCode clazz detail + id <- getWord16be token <- getToken tokenLength options <- getOptions payload <- getPayload - return (Message { messageHeader = header - , messageToken = token - , messageOptions = options - , messagePayload = payload }) + return Message { messageVersion = version + , messageType = msgType + , messageCode = code + , messageId = fromIntegral id + , messageToken = token + , messageOptions = options + , messagePayload = payload } -- | Decode a CoAP message according to the specification. decode :: BS.ByteString -> Message @@ -234,12 +210,12 @@ let (responseClass, responseDetail) = encodeResponseCode detail in (.|.) (shiftL responseClass 5) responseDetail -putHeader :: MessageHeader -> Word8 -> Put -putHeader header tokenLength = do - let version = fromIntegral (messageVersion header) :: Word8 - let eType = encodeType (messageType header) - let code = messageCode header - let id = messageId header +putHeader :: Message -> Word8 -> Put +putHeader message tokenLength = do + let version = fromIntegral (messageVersion message) :: Word8 + let eType = encodeType (messageType message) + let code = messageCode message + let id = messageId message putWord8 ((.|.) ((.|.) (shiftL version 6) (shiftL eType 4)) ((.&.) tokenLength 0x0F)) putWord8 (encodeCode code) putWord16be id @@ -250,12 +226,9 @@ putOptionInt :: Int -> Put putOptionInt 0 = return () -putOptionInt n = do - if n < 256 - then putWord8 (fromIntegral n) - else if n < 65536 - then putWord16be (fromIntegral n) - else putWord32be (fromIntegral n) +putOptionInt n | n < 256 = putWord8 (fromIntegral n) + | n < 65536 = putWord16be (fromIntegral n) + | otherwise = putWord32be (fromIntegral n) encodeOptionInt :: Int -> BS.ByteString encodeOptionInt n = BS.pack (unpack (runPut (putOptionInt n))) @@ -336,7 +309,7 @@ putMessage msg = do let token = messageToken msg let tokenLength = BS.length token - putHeader (messageHeader msg) (fromIntegral tokenLength) + putHeader msg (fromIntegral tokenLength) putToken token putOptions (messageOptions msg) putPayload (messagePayload msg) diff --git a/src/Network/CoAP/Messaging.hs b/src/Network/CoAP/Messaging.hs --- a/src/Network/CoAP/Messaging.hs +++ b/src/Network/CoAP/Messaging.hs @@ -1,7 +1,8 @@ module Network.CoAP.Messaging ( createMessagingState , MessagingState -, messagingLoop +, startMessaging +, stopMessaging , recvRequest , sendResponse , sendRequest @@ -16,6 +17,8 @@ import Data.ByteString (empty) import Control.Concurrent.STM import Control.Concurrent +import Control.Exception +import Control.Concurrent.Async import Data.Maybe import Data.Time import System.Random @@ -28,7 +31,7 @@ , retransmitCount :: Integer } deriving (Show) cmpMessageState a b = - messageId (messageHeader (message (messageContext a))) == messageId (messageHeader (message (messageContext b))) && + messageId (message (messageContext a)) == messageId (message (messageContext b)) && srcEndpoint (messageContext a) == srcEndpoint (messageContext b) instance Eq MessageState where @@ -122,7 +125,7 @@ takeMessageByToken token = takeMessageMatching (\x -> token == messageToken (message (messageContext x))) takeMessageByIdAndOrigin :: MessageId -> Endpoint -> TVar MessageList -> STM (Maybe MessageState) -takeMessageByIdAndOrigin msgId origin = takeMessageMatching (\x -> (origin == dstEndpoint (messageContext x)) && (msgId == messageId (messageHeader (message (messageContext x))))) +takeMessageByIdAndOrigin msgId origin = takeMessageMatching (\x -> (origin == dstEndpoint (messageContext x)) && (msgId == messageId (message (messageContext x)))) recvLoop :: MessagingState -> IO () recvLoop state@(MessagingState transport store) = do @@ -138,9 +141,9 @@ , insertionTime = now , replyTimeout = 0 , retransmitCount = 0 } - let msgId = messageId (messageHeader message) - let msgType = messageType (messageHeader message) - let msgCode = messageCode (messageHeader message) + let msgId = messageId message + let msgType = messageType message + let msgCode = messageCode message atomically (when (msgType == ACK) (do _ <- takeMessageByIdAndOrigin msgId srcEndpoint (unconfirmedMessages store) return ())) atomically (when (msgCode /= CodeEmpty) (queueMessage messageState (incomingMessages store))) @@ -150,7 +153,7 @@ sendLoop state@(MessagingState transport store) = do msgState <- atomically (do msgState <- takeMessageRetry (outgoingMessages store) - let msgType = messageType (messageHeader (message (messageContext msgState))) + let msgType = messageType (message (messageContext msgState)) when (msgType == CON) (queueMessage msgState (unconfirmedMessages store)) return msgState) let msgCtx = messageContext msgState @@ -164,12 +167,11 @@ createAckMessage now origMessageState = let origCtx = messageContext origMessageState origMessage = message origCtx - origHeader = messageHeader origMessage - newHeader = MessageHeader { messageVersion = messageVersion origHeader - , messageType = ACK - , messageCode = CodeEmpty - , messageId = messageId origHeader } - newMessage = Message { messageHeader = newHeader + + newMessage = Message { messageVersion = messageVersion origMessage + , messageType = ACK + , messageCode = CodeEmpty + , messageId = messageId origMessage , messageToken = empty , messageOptions = [] , messagePayload = Nothing } @@ -229,13 +231,17 @@ retransmitLoop state -messagingLoop :: MessagingState -> IO () -messagingLoop state = do - recvThread <- forkIO (recvLoop state) - sendThread <- forkIO (sendLoop state) - ackThread <- forkIO (ackLoop state) - retransmitLoop state +runLoop :: MessagingState -> (MessagingState -> IO ()) -> IO () +runLoop state fn = do + err <- try (fn state) :: IO (Either AsyncException ()) + return () +startMessaging :: MessagingState -> IO [Async ()] +startMessaging state = mapM (async . runLoop state) [recvLoop, sendLoop, ackLoop, retransmitLoop] + +stopMessaging :: MessagingState -> [Async ()] -> IO () +stopMessaging state = mapM_ cancel + sendMessage :: Message -> Endpoint -> MessagingState -> IO () sendMessage message dstEndpoint (MessagingState transport store) = do srcEndpoint <- localEndpoint transport @@ -255,7 +261,7 @@ recvMessageMatching matchFilter (MessagingState _ store) = do msgState <- atomically (takeMessageRetryMatching matchFilter (incomingMessages store)) let msgCtx = messageContext msgState - let msgType = messageType (messageHeader (message msgCtx)) + let msgType = messageType (message msgCtx) when (msgType == CON) (atomically (queueMessage msgState (unackedMessages store))) return msgCtx @@ -267,31 +273,27 @@ cmpMessageCode _ _ = False recvRequest :: MessagingState -> IO MessageContext -recvRequest = recvMessageMatching (cmpMessageCode (CodeRequest GET) . messageCode . messageHeader . message . messageContext) +recvRequest = recvMessageMatching (cmpMessageCode (CodeRequest GET) . messageCode . message . messageContext) createAckResponse :: Message -> Message createAckResponse response = - let origHeader = messageHeader response - header = MessageHeader { messageVersion = messageVersion origHeader - , messageType = ACK - , messageCode = messageCode origHeader - , messageId = messageId origHeader } - in Message { messageHeader = header - , messageToken = messageToken response - , messageOptions = messageOptions response - , messagePayload = messagePayload response } + Message { messageVersion = messageVersion response + , messageType = ACK + , messageCode = messageCode response + , messageId = messageId response + , messageToken = messageToken response + , messageOptions = messageOptions response + , messagePayload = messagePayload response } setMessageId :: MessageId -> Message -> Message setMessageId msgId response = - let origHeader = messageHeader response - header = MessageHeader { messageVersion = messageVersion origHeader - , messageType = messageType origHeader - , messageCode = messageCode origHeader - , messageId = msgId } - in Message { messageHeader = header - , messageToken = messageToken response - , messageOptions = messageOptions response - , messagePayload = messagePayload response } + Message { messageVersion = messageVersion response + , messageType = messageType response + , messageCode = messageCode response + , messageId = msgId + , messageToken = messageToken response + , messageOptions = messageOptions response + , messagePayload = messagePayload response } allocateMessageId :: IO MessageId allocateMessageId = randomIO @@ -305,7 +307,7 @@ msgId <- case unackedMsg of Nothing -> allocateMessageId - _ -> return (messageId (messageHeader (message requestCtx))) + _ -> return (messageId (message requestCtx)) let outgoingMessage = case unackedMsg of Nothing -> setMessageId msgId response @@ -314,13 +316,12 @@ sendMessage outgoingMessage origin state sendRequest :: Message -> Endpoint -> MessagingState -> IO () -sendRequest (Message (MessageHeader msgVersion msgType msgCode _) tkn opts payload) msgdest state = do +sendRequest (Message msgVersion msgType msgCode _ tkn opts payload) msgdest state = do msgId <- allocateMessageId - let msgHeader = MessageHeader { messageVersion = msgVersion - , messageType = msgType - , messageCode = msgCode - , messageId = msgId } - let msg = Message { messageHeader = msgHeader + let msg = Message { messageVersion = msgVersion + , messageType = msgType + , messageCode = msgCode + , messageId = msgId , messageToken = tkn , messageOptions = opts , messagePayload = payload } @@ -328,5 +329,5 @@ recvResponse :: Message -> Endpoint -> MessagingState -> IO MessageContext recvResponse reqMessage endpoint = recvMessageMatching matchFilter - where matchFilter x = cmpMessageCode (CodeResponse Created) (messageCode (messageHeader (message (messageContext x)))) && messageToken reqMessage == messageToken (message (messageContext x)) + where matchFilter x = cmpMessageCode (CodeResponse Created) (messageCode (message (messageContext x))) && messageToken reqMessage == messageToken (message (messageContext x)) diff --git a/src/Network/CoAP/Server.hs b/src/Network/CoAP/Server.hs --- a/src/Network/CoAP/Server.hs +++ b/src/Network/CoAP/Server.hs @@ -22,7 +22,6 @@ , Option(..) , MediaType(..) , createServer -, shutdownServer , Server(..) , RequestHandler ) where @@ -39,29 +38,31 @@ -- | A CoAP server instance. data Server = Server { runServer :: IO () - , msgThreadId :: Async () } + , stopServer :: IO () } -- | Create a CoAP server with a given transport and request handler createServer :: Transport -> RequestHandler -> IO Server createServer transport handler = do state <- createMessagingState transport - msgThread <- async (messagingLoop state) + msgThreads <- startMessaging state return Server { runServer = requestLoop state handler - , msgThreadId = msgThread } + , stopServer = shutdownServer state msgThreads } -- | Shutdown a CoAP server. -shutdownServer :: Server -> IO () -shutdownServer server = wait (msgThreadId server) -- TODO: Stop messaging thread. +shutdownServer :: MessagingState -> [Async ()] -> IO () +shutdownServer state threads = do + stopMessaging state threads + mapM_ wait threads createRequest :: MessageContext -> Request createRequest reqCtx = let msg = message reqCtx - (CodeRequest method) = messageCode (messageHeader msg) + (CodeRequest method) = messageCode msg in Request { requestMethod = method , requestOptions = messageOptions msg , requestPayload = messagePayload msg - , requestReliable = messageType (messageHeader msg) == CON } + , requestReliable = messageType msg == CON } handleRequest :: MessageContext -> RequestHandler -> MessagingState -> IO () handleRequest requestCtx requestHandler state = do @@ -82,13 +83,11 @@ createResponseMessage :: Message -> Response -> Message createResponseMessage origMsg response = - let origHeader = messageHeader origMsg - header = MessageHeader { messageVersion = messageVersion origHeader - , messageType = messageType origHeader - , messageCode = CodeResponse (responseCode response) - , messageId = messageId origHeader } - in Message { messageHeader = header - , messageToken = messageToken origMsg - , messageOptions = responseOptions response - , messagePayload = responsePayload response } + Message { messageVersion = messageVersion origMsg + , messageType = messageType origMsg + , messageCode = CodeResponse (responseCode response) + , messageId = messageId origMsg + , messageToken = messageToken origMsg + , messageOptions = responseOptions response + , messagePayload = responsePayload response } diff --git a/src/Network/CoAP/Types.hs b/src/Network/CoAP/Types.hs --- a/src/Network/CoAP/Types.hs +++ b/src/Network/CoAP/Types.hs @@ -107,14 +107,6 @@ -- | Valid CoAP message types. data MessageType = CON | NON | ACK | RST deriving (Show, Eq) --- | CoAP message header. -data MessageHeader = MessageHeader - { messageVersion :: MessageVersion - , messageType :: MessageType - , messageCode :: MessageCode - , messageId :: MessageId - } deriving (Show) - -- | CoAP version. Only a value of 1 is valid. type MessageVersion = Int -- | CoAP message id. This is unique for each message from a given endpoint. @@ -124,7 +116,10 @@ -- | CoAP message record. Can be encoded and decoded to ByteString. data Message = Message - { messageHeader :: MessageHeader + { messageVersion :: MessageVersion + , messageType :: MessageType + , messageCode :: MessageCode + , messageId :: MessageId , messageToken :: Token , messageOptions :: [Option] , messagePayload :: Maybe Payload diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -8,7 +8,7 @@ import Test.QuickCheck import Prelude hiding (null, length) import Control.Concurrent.Chan -import Control.Concurrent +import Control.Concurrent.Async import Control.Monad import System.Random import System.Timeout @@ -54,7 +54,7 @@ let transportA = createUnstableTransport linkAProb (endpointA, chanA) (endpointB, chanB) let transportB = createUnstableTransport linkBProb (endpointB, chanB) (endpointA, chanA) server <- S.createServer transportA testHandler - serverThread <- forkIO (S.runServer server) + serverThread <- async (S.runServer server) client <- C.createClient transportB reqs <- generate (vector 10) mapM_ (\req -> do @@ -64,7 +64,8 @@ putStrLn "Got response, checking" assertEqual "Bad response code" C.Created (C.responseCode res) assertEqual "Bad payload" "Hello, Client" (unpack (fromJust (C.responsePayload res))) - ) reqs) + ) reqs + S.stopServer server) -- TODO, negative tests: -- * Invalid token length @@ -78,17 +79,6 @@ instance Arbitrary Method where arbitrary = elements [PUT, GET, POST, DELETE] -instance Arbitrary MessageHeader where - arbitrary = do - msgType <- arbitrary - msgMethod <- arbitrary - msgId <- arbitrary - return (MessageHeader { messageVersion = 1 - , messageType = msgType - , messageCode = CodeRequest msgMethod - , messageId = msgId}) - - instance Arbitrary ByteString where arbitrary = suchThat (fmap pack arbitrary) (\s -> ((length s > 0) && (length s <= 8))) @@ -111,11 +101,16 @@ instance Arbitrary Message where arbitrary = do - hdr <- arbitrary + msgType <- arbitrary + msgMethod <- arbitrary + msgId <- arbitrary tkn <- arbitrary options <- arbitrary payload <- arbitrary - return (Message { messageHeader = hdr + return (Message { messageVersion = 1 + , messageType = msgType + , messageCode = CodeRequest msgMethod + , messageId = msgId , messageToken = tkn , messageOptions = [options] , messagePayload = payload })