hcoap (empty) → 0.1.0.0
raw patch · 13 files changed
+1300/−0 lines, 13 filesdep +HUnitdep +QuickCheckdep +asyncsetup-changed
Dependencies added: HUnit, QuickCheck, async, base, binary, bytestring, hcoap, mtl, network, random, stm, time
Files
- LICENSE +30/−0
- Setup.hs +2/−0
- example-client/Main.hs +21/−0
- example-server/Main.hs +28/−0
- hcoap.cabal +67/−0
- src/Network/CoAP/Client.hs +75/−0
- src/Network/CoAP/Message.hs +26/−0
- src/Network/CoAP/MessageCodec.hs +346/−0
- src/Network/CoAP/Messaging.hs +332/−0
- src/Network/CoAP/Server.hs +86/−0
- src/Network/CoAP/Transport.hs +25/−0
- src/Network/CoAP/Types.hs +126/−0
- test/Spec.hs +136/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2016, Ulf Lilleengen++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 Ulf Lilleengen 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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ example-client/Main.hs view
@@ -0,0 +1,21 @@+import Network.CoAP.Client+import Network.CoAP.Transport+import Network.Socket++import qualified Data.ByteString.Char8 as B++main :: IO ()+main = do+ let request = Request { requestMethod = GET+ , requestOptions = [UriPath (B.pack "hello")]+ , requestPayload = Nothing+ , requestReliable = True }+ withSocketsDo $ do+ sock <- socket AF_INET6 Datagram defaultProtocol+ bindSocket sock (SockAddrInet6 12345 0 iN6ADDR_ANY 0)+ let transport = createUDPTransport sock+ let dest = SockAddrInet6 5683 0 (0, 0, 0, 0) 0+ client <- createClient transport+ response <- doRequest client dest request+ putStrLn ("Got response: " ++ show response)+ return ()
+ example-server/Main.hs view
@@ -0,0 +1,28 @@+import Network.CoAP.Server+import Network.CoAP.Transport+import Network.Socket+import qualified Data.ByteString.Char8 as B++findPath :: [Option] -> B.ByteString+findPath [] = B.empty+findPath (option:options) =+ case option of+ UriPath value -> value+ _ -> findPath options++requestHandler :: RequestHandler+requestHandler (request, _) = do+ let options = requestOptions request+ let path = findPath options+ let payload = Just (B.pack ("{\"path\":\"" ++ B.unpack path ++ "\"}"))+ return (Response Content [ContentFormat ApplicationJson] payload)+ ++main :: IO ()+main =+ withSocketsDo $ do+ sock <- socket AF_INET6 Datagram defaultProtocol+ bindSocket sock (SockAddrInet6 12345 0 iN6ADDR_ANY 0)+ server <- createServer (createUDPTransport sock) requestHandler+ runServer server+
+ hcoap.cabal view
@@ -0,0 +1,67 @@+name: hcoap+version: 0.1.0.0+synopsis: Library for handling CoAP protocol+license: BSD3+license-file: LICENSE+author: Ulf Lilleengen+maintainer: ulf.lilleengen@gmail.com+category: Network+build-type: Simple+homepage: https://github.com/lulf/hcoap+description: CoAP library for writing CoAP clients, servers or just for decoding and encoding+ CoAP messages. The Network.CoAP.Server and Network.CoAP.Client modules allows+ building CoAP servers and clients on top of a messaging layer which provides+ reliable transport of CoAP requests/responses.+synopsis: CoAP implementation for Haskell.+stability: experimental+cabal-version: >=1.10++library+ exposed-modules: Network.CoAP.Server, Network.CoAP.Client, Network.CoAP.Message, Network.CoAP.Transport+ other-modules: Network.CoAP.Types, Network.CoAP.MessageCodec, Network.CoAP.Messaging++ build-depends: base >= 4.8 && < 4.9,+ bytestring >= 0.10 && < 0.11,+ binary >= 0.7 && < 0.8,+ network >= 2.6 && < 2.7,+ stm >= 2.4 && < 2.5,+ mtl >= 2.2 && < 2.3,+ time >= 1.5 && < 1.6,+ random >= 1.1 && < 1.2,+ async >= 2.1 && < 2.2+ hs-source-dirs: src+ default-language: Haskell2010++test-suite hcoap-test+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ build-depends: base,+ hcoap,+ network,+ QuickCheck,+ HUnit,+ random,+ bytestring+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ default-language: Haskell2010++executable hcoap-example-server+ hs-source-dirs: example-server+ main-is: Main.hs+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends: base,+ network,+ bytestring,+ hcoap+ default-language: Haskell2010++executable hcoap-example-client+ hs-source-dirs: example-client+ main-is: Main.hs+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends: base,+ network,+ bytestring,+ hcoap+ default-language: Haskell2010
+ src/Network/CoAP/Client.hs view
@@ -0,0 +1,75 @@+{-|+Module: Network.CoAP.Client+Description: CoAP client library+Maintainer: ulf.lilleengen@gmail.com+License: BSD3++The CoAP client API is intended to provide the minimal building block needed for sending CoAP requests. The API exposes CoAP request and response types and handles all internal messaging details of the CoAP protocol.+-}+module Network.CoAP.Client+( Request(..)+, Client+, Method(..)+, Response(..)+, ResponseCode(..)+, Option(..)+, OptionString+, MediaType(..)+, doRequest+, createClient+, shutdownClient+) where++import Network.CoAP.Messaging+import Network.CoAP.Types+import Control.Monad.State+import Control.Concurrent.Async+import Control.Concurrent.STM+import Network.Socket+import Data.ByteString+import Data.Word+import System.Random+ +-- | 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 ()}++-- | 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)+ return Client { doRequest = doRequestInternal state+ , msgThreadId = msgThread }++-- | Shuts down the internal messaging threads and stops the client+shutdownClient :: Client -> IO ()+shutdownClient client = wait (msgThreadId client) -- TODO: Actually shut down the threads...++generateToken :: Int -> IO [Word8]+generateToken 0 = return []+generateToken len = do+ tkn <- randomIO+ next <- generateToken (len - 1)+ return (tkn:next)++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 }+ sendRequest msg dest state+ responseCtx <- recvResponse msg dest state+ let (Message (MessageHeader _ _ (CodeResponse rCode) _) _ opts pload) = message responseCtx+ return Response { responseCode = rCode+ , responseOptions = opts+ , responsePayload = pload }
+ src/Network/CoAP/Message.hs view
@@ -0,0 +1,26 @@+{-|+Module: Network.CoAP.Message+Description: CoAP message library+Maintainer: ulf.lilleengen@gmail.com+License: BSD3++The CoAP message API is intended to provide the minimal building block needed for creating, encoding and decoding CoAP messages. The module can be used to build alternative CoAP messaging layers.+-}+module Network.CoAP.Message+( Message(..)+, MessageHeader(..)+, MessageVersion(..)+, MessageType(..)+, MessageId(..)+, MessageCode(..)+, Token(..)+, Payload(..)+, Option(..)+, MediaType(..)+, Method(..)+, encode+, decode+) where++import Network.CoAP.Types+import Network.CoAP.MessageCodec (encode, decode)
+ src/Network/CoAP/MessageCodec.hs view
@@ -0,0 +1,346 @@+module Network.CoAP.MessageCodec+( decode+, encode+) where++import Debug.Trace+import Network.CoAP.Types+import Data.ByteString.Lazy+import qualified Data.ByteString as BS+import Data.Word+import Data.Maybe+import Data.List (sortOn)+import Data.Binary hiding (encode, decode)+import qualified Data.Binary as DB+import Data.Binary.Get+import Data.Binary.Put+import Data.Bits+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 _ = fail "Unknown type"++getRequestMethod :: Word8 -> Get Method+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)++getCode :: Word8 -> Word8 -> Get MessageCode+getCode 0 0 = return (CodeEmpty)+getCode 0 detail = do+ method <- getRequestMethod detail+ return (CodeRequest method)++getCode code detail =+ if code == 2 || code == 4 || code == 5+ then do+ responseCode <- getResponseCode detail+ return (CodeResponse responseCode)+ else fail "Unknown class"+++getTokenLength :: Word8 -> Get Word8+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)++getToken :: Word8 -> Get (Token)+getToken 0 = return BS.empty+getToken n = do+ str <- getByteString (fromIntegral n)+ return str++intToMediaType 0 = TextPlain+intToMediaType 40 = ApplicationLinkFormat+intToMediaType 41 = ApplicationXml+intToMediaType 42 = ApplicationOctetStream+intToMediaType 47 = ApplicationExi+intToMediaType 50 = ApplicationJson++decodeOption :: Int -> BS.ByteString -> Int -> Option+decodeOption 1 value _ = IfMatch value+decodeOption 3 value _ = UriHost value+decodeOption 4 value _ = ETag value+decodeOption 5 _ _ = IfNoneMatch+decodeOption 7 value valueLen = UriPort (decodeOptionInt value valueLen)+decodeOption 8 value _ = LocationPath value+decodeOption 11 value _ = UriPath value+decodeOption 12 value valueLen = ContentFormat (intToMediaType (decodeOptionInt value valueLen))+decodeOption 14 value valueLen = MaxAge (decodeOptionInt value valueLen)+decodeOption 15 value _ = UriQuery value+decodeOption 17 value valueLen = Accept (decodeOptionInt value valueLen)+decodeOption 20 value _ = LocationQuery value+decodeOption 35 value _ = ProxyUri value+decodeOption 39 value _ = ProxyScheme value+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)++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++getOption :: Int -> Get (Maybe Option)+getOption lastCode = do+ e <- isEmpty+ if e+ then return Nothing+ else do+ tmp <- getWord8+ if tmp == 0xFF+ then return Nothing+ else do+ let delta = fromIntegral ((.&.) (shiftR tmp 4) 0x0F)+ let valueLength = fromIntegral ((.&.) tmp 0x0F)+ fullDelta <- getOptionPart delta+ let optCode = lastCode + fullDelta+ fullLength <- getOptionPart valueLength+ value <- getByteString fullLength+ let option = decodeOption optCode value fullLength+ return (Just option)++getOptionsLoop :: Int -> Get ([Option])+getOptionsLoop lastCode = do+ opt <- getOption lastCode+ case opt of+ Nothing -> do+ return []+ Just o -> do+ let optCode = fst (encodeOption o)+ opts <- getOptionsLoop optCode+ return (o:opts)++getOptions :: Get ([Option])+getOptions = getOptionsLoop 0++getPayload :: Get (Maybe Payload)+getPayload = do+ str <- getRemainingLazyByteString+ return (if null str+ then Nothing+ else Just (BS.pack (unpack str)))+++getMessage :: Get Message+getMessage = do+ (header, tokenLength) <- getHeader+ token <- getToken tokenLength+ options <- getOptions+ payload <- getPayload+ return (Message { messageHeader = header+ , messageToken = token+ , messageOptions = options+ , messagePayload = payload })++-- | Decode a CoAP message according to the specification.+decode :: BS.ByteString -> Message+decode msg = runGet getMessage (fromStrict msg)++encodeType :: MessageType -> Word8+encodeType CON = 0+encodeType NON = 1+encodeType ACK = 2+encodeType RST = 3++encodeRequestMethod :: Method -> Word8+encodeRequestMethod GET = 1+encodeRequestMethod POST = 2+encodeRequestMethod PUT = 3+encodeRequestMethod DELETE = 4++encodeResponseCode :: ResponseCode -> (Word8, Word8)+encodeResponseCode Created = (2, 1)+encodeResponseCode Deleted = (2, 2)+encodeResponseCode Valid = (2, 3)+encodeResponseCode Changed = (2, 4)+encodeResponseCode Content = (2, 5)+encodeResponseCode BadRequest = (4, 0)+encodeResponseCode Unauthorized = (4, 1)+encodeResponseCode BadOption = (4, 2)+encodeResponseCode Forbidden = (4, 3)+encodeResponseCode NotFound = (4, 4)+encodeResponseCode MethodNotAllowed = (4, 5)+encodeResponseCode NotAcceptable = (4, 6)+encodeResponseCode PreconditionFailed = (4, 12)+encodeResponseCode RequestEntityTooLarge = (4, 13)+encodeResponseCode UnsupportedFormat = (4, 15)+encodeResponseCode InternalServerError = (5, 0)+encodeResponseCode NotImplemented = (5, 1)+encodeResponseCode BadGateway = (5, 2)+encodeResponseCode ServiceUnavailable = (5, 3)+encodeResponseCode GatewayTimeout = (5, 4)+encodeResponseCode ProxyingNotSupported = (5, 5)++encodeCode :: MessageCode -> Word8+encodeCode CodeEmpty = 0+encodeCode (CodeRequest detail) = encodeRequestMethod detail+encodeCode (CodeResponse detail) =+ 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+ putWord8 ((.|.) ((.|.) (shiftL version 6) (shiftL eType 4)) ((.&.) tokenLength 0x0F))+ putWord8 (encodeCode code)+ putWord16be id+++putToken :: Token -> Put+putToken = putByteString++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)++encodeOptionInt :: Int -> BS.ByteString+encodeOptionInt n = BS.pack (unpack (runPut (putOptionInt n)))++mediaTypeToInt :: MediaType -> Int+mediaTypeToInt TextPlain = 0+mediaTypeToInt ApplicationLinkFormat = 40+mediaTypeToInt ApplicationXml = 41+mediaTypeToInt ApplicationOctetStream = 42+mediaTypeToInt ApplicationExi = 47+mediaTypeToInt ApplicationJson = 50++encodeOption :: Option -> (Int, BS.ByteString)+encodeOption (IfMatch value) = (1, value)+encodeOption (UriHost value) = (3, value)+encodeOption (ETag value) = (4, value)+encodeOption IfNoneMatch = (5, BS.empty)+encodeOption (UriPort value) = (7, encodeOptionInt value)+encodeOption (LocationPath value) = (8, value)+encodeOption (UriPath value) = (11, value)+encodeOption (ContentFormat mediaType) = (12, encodeOptionInt (mediaTypeToInt mediaType))+encodeOption (MaxAge value) = (14, encodeOptionInt value)+encodeOption (UriQuery value) = (15, value)+encodeOption (Accept value) = (17, encodeOptionInt value)+encodeOption (LocationQuery value) = (20, value)+encodeOption (ProxyUri value) = (35, value)+encodeOption (ProxyScheme value) = (39, value)+encodeOption (Size1 value) = (60, encodeOptionInt value)++encodeOptionPart :: Int -> Word8+encodeOptionPart value =+ if value > 268+ then 14+ else if value > 12+ then 13+ else fromIntegral value++putOptionPart :: Word8 -> Int -> Put+putOptionPart valueLen value = do+ if valueLen == 13+ then putWord8 (fromIntegral (value - 13))+ else if valueLen == 14+ then putWord16be (fromIntegral (value - 269))+ else return ()++putOption :: Int -> BS.ByteString -> Put+putOption nextDelta optValue = do+ let optValueLen = BS.length optValue+ let valueLen = encodeOptionPart optValueLen+ let deltaValue = encodeOptionPart nextDelta+ let tmp = (.|.) (shiftL deltaValue 4) ((.&.) valueLen 0x0F)++ putWord8 tmp+ putOptionPart deltaValue nextDelta+ putOptionPart valueLen optValueLen+ putByteString optValue++putOptionsLoop :: Int -> [Option] -> Put+putOptionsLoop lastNumber [] = return ()+putOptionsLoop lastNumber (opt:options) = do+ let (optNumber, optValue) = encodeOption opt+ let nextDelta = optNumber - lastNumber+ putOption nextDelta optValue+ putOptionsLoop optNumber options++putOptions :: [Option] -> Put+putOptions options = do+ let sortedOptions = sortOn (\o -> fst (encodeOption o)) options+ putOptionsLoop 0 sortedOptions++putPayload :: Maybe Payload -> Put+putPayload Nothing = return ()+putPayload (Just payload) = do+ putWord8 0xFF+ putByteString payload++putMessage :: Message -> Put+putMessage msg = do+ let token = messageToken msg+ let tokenLength = BS.length token+ putHeader (messageHeader msg) (fromIntegral tokenLength)+ putToken token+ putOptions (messageOptions msg)+ putPayload (messagePayload msg)++-- | Encode a CoAP message according to the specification.+encode :: Message -> BS.ByteString+encode msg = toStrict (runPut (putMessage msg))
+ src/Network/CoAP/Messaging.hs view
@@ -0,0 +1,332 @@+module Network.CoAP.Messaging+( createMessagingState+, MessagingState+, messagingLoop+, recvRequest+, sendResponse+, sendRequest+, recvResponse+) where+import Network.CoAP.Types+import Network.CoAP.MessageCodec+import Data.List (deleteBy, find, partition, filter, minimumBy, delete, sortBy)+import Control.Monad+import Control.Monad.State.Lazy+import Data.Ord+import Data.ByteString (empty)+import Control.Concurrent.STM+import Control.Concurrent+import Data.Maybe+import Data.Time+import System.Random++type TimeStamp = UTCTime++data MessageState = MessageState { messageContext :: MessageContext+ , insertionTime :: TimeStamp+ , replyTimeout :: Double + , retransmitCount :: Integer } deriving (Show)++cmpMessageState a b =+ messageId (messageHeader (message (messageContext a))) == messageId (messageHeader (message (messageContext b))) &&+ srcEndpoint (messageContext a) == srcEndpoint (messageContext b)++instance Eq MessageState where+ (==) = cmpMessageState++-- A message store contains an inbound and outbound list of messages that needs to be ACKed+type MessageList = [MessageState]+data MessagingStore = MessagingStore { incomingMessages :: TVar MessageList+ , outgoingMessages :: TVar MessageList+ , unackedMessages :: TVar MessageList + , unconfirmedMessages :: TVar MessageList }+data MessagingState = MessagingState Transport MessagingStore++createMessagingState :: Transport -> IO MessagingState+createMessagingState transport = do+ incoming <- newTVarIO []+ outgoing <- newTVarIO []+ unacked <- newTVarIO []+ unconfirmed <- newTVarIO []+ return (MessagingState transport (MessagingStore incoming outgoing unacked unconfirmed))++queueMessages :: [MessageState] -> TVar MessageList -> STM ()+queueMessages messages msgListVar = do+ msgList <- readTVar msgListVar+ writeTVar msgListVar (messages ++ msgList)++queueMessage :: MessageState -> TVar MessageList -> STM ()+queueMessage message = queueMessages [message]++takeMessageRetry :: TVar MessageList -> STM MessageState+takeMessageRetry msgListVar = do+ msgList <- readTVar msgListVar+ if null msgList+ then retry+ else do+ let message = minimumBy (comparing insertionTime) msgList+ let newMsgList = delete message msgList+ writeTVar msgListVar newMsgList+ return message++takeMessagesRetryMatching :: (MessageState -> Bool) -> TVar MessageList -> STM [MessageState]+takeMessagesRetryMatching msgFilter msgListVar = do+ msgList <- readTVar msgListVar+ if null msgList+ then retry+ else do+ let (matchedMessages, remainingMessages) = partition msgFilter msgList+ writeTVar msgListVar remainingMessages+ return matchedMessages++takeMessagesOlderThan :: TimeStamp -> TVar MessageList -> STM [MessageState]+takeMessagesOlderThan timeStamp = takeMessagesRetryMatching (\x -> timeStamp > insertionTime x)++checkRetransmit :: TimeStamp -> MessageState -> Bool+checkRetransmit now msgState =+ let timeout = replyTimeout msgState+ startTime = insertionTime msgState+ endTime = addUTCTime (realToFrac timeout) startTime+ in now > endTime++takeMessagesToRetransmit :: TimeStamp -> TVar MessageList -> STM [MessageState]+takeMessagesToRetransmit now = takeMessagesRetryMatching (checkRetransmit now)++takeMessageRetryMatching :: (MessageState -> Bool) -> TVar MessageList -> STM MessageState+takeMessageRetryMatching matchFilter msgListVar = do+ msgList <- readTVar msgListVar+ if null msgList+ then retry+ else do+ let sortedMsgList = sortBy (comparing insertionTime) msgList+ let msg = find matchFilter sortedMsgList+ case msg of+ Nothing -> retry+ Just m -> do+ let newMsgList = delete m msgList+ writeTVar msgListVar newMsgList+ return m++takeMessageMatching :: (MessageState -> Bool) -> TVar MessageList -> STM (Maybe MessageState)+takeMessageMatching matchFilter msgListVar = do+ msgList <- readTVar msgListVar+ if null msgList+ then return Nothing+ else do+ let sortedMsgList = sortBy (comparing insertionTime) msgList+ let (foundMessages, remainingMessages) = partition matchFilter sortedMsgList+ writeTVar msgListVar remainingMessages+ return (listToMaybe foundMessages)++takeMessageByToken :: Token -> TVar MessageList -> STM (Maybe MessageState)+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)))))++recvLoop :: MessagingState -> IO ()+recvLoop state@(MessagingState transport store) = do+ {-putStrLn "Waiting for UDP packet"-}+ (msgData, srcEndpoint) <- recvFrom transport+ dstEndpoint <- localEndpoint transport+ now <- getCurrentTime+ let message = decode msgData+ let messageCtx = MessageContext { message = message+ , srcEndpoint = srcEndpoint+ , dstEndpoint = dstEndpoint }+ let messageState = MessageState { messageContext = messageCtx + , insertionTime = now+ , replyTimeout = 0+ , retransmitCount = 0 }+ let msgId = messageId (messageHeader message)+ let msgType = messageType (messageHeader message)+ let msgCode = messageCode (messageHeader message)+ atomically (when (msgType == ACK) (do _ <- takeMessageByIdAndOrigin msgId srcEndpoint (unconfirmedMessages store)+ return ()))+ atomically (when (msgCode /= CodeEmpty) (queueMessage messageState (incomingMessages store)))+ recvLoop state+ +sendLoop :: MessagingState -> IO ()+sendLoop state@(MessagingState transport store) = do+ msgState <- atomically (do+ msgState <- takeMessageRetry (outgoingMessages store)+ let msgType = messageType (messageHeader (message (messageContext msgState)))+ when (msgType == CON) (queueMessage msgState (unconfirmedMessages store))+ return msgState)+ let msgCtx = messageContext msgState+ let msgData = encode (message msgCtx)+ let origin = dstEndpoint msgCtx+ _ <- sendTo transport msgData origin+ sendLoop state+++createAckMessage :: UTCTime -> MessageState -> MessageState+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+ , messageToken = empty+ , messageOptions = []+ , messagePayload = Nothing }+ newCtx = MessageContext { message = newMessage+ , srcEndpoint = dstEndpoint origCtx+ , dstEndpoint = srcEndpoint origCtx}+ in MessageState { messageContext = newCtx+ , replyTimeout = 0+ , retransmitCount = 0+ , insertionTime = now }++ ++ackTimeout :: Double+ackTimeout = 2++ackLoop :: MessagingState -> IO ()+ackLoop state@(MessagingState _ store) = do+ takeStamp <- getCurrentTime+ let nomTime = realToFrac (-ackTimeout)+ let oldestTime = addUTCTime nomTime takeStamp+ oldMessages <- atomically (takeMessagesOlderThan oldestTime (unackedMessages store))+ if null oldMessages+ then do+ threadDelay 100000+ ackLoop state+ else do+ putStrLn "Timeout! Queueing ack messages"+ now <- getCurrentTime+ let ackMessages = map (createAckMessage now) oldMessages+ atomically (queueMessages ackMessages (outgoingMessages store))+ ackLoop state++ackRandomFactor :: Double+ackRandomFactor = 1.5++maxRetransmitCount :: Integer+maxRetransmitCount = 4++adjustRetransmissionState :: TimeStamp -> MessageState -> MessageState+adjustRetransmissionState now msgState =+ MessageState { messageContext = messageContext msgState+ , insertionTime = now+ , replyTimeout = replyTimeout msgState * 2+ , retransmitCount = retransmitCount msgState + 1 }++retransmitLoop :: MessagingState -> IO ()+retransmitLoop state@(MessagingState _ store) = do+ now <- getCurrentTime+ toRetransmit <- atomically (takeMessagesToRetransmit now (unconfirmedMessages store))+ if null toRetransmit+ then threadDelay 100000+ else (do+ putStrLn ("Attempting to retransmit messages " ++ show toRetransmit)+ let adjustedMessages = filter (\s -> retransmitCount s <= maxRetransmitCount) (map (adjustRetransmissionState now) toRetransmit)+ atomically (queueMessages adjustedMessages (outgoingMessages store)))+ retransmitLoop state+ ++messagingLoop :: MessagingState -> IO ()+messagingLoop state = do+ recvThread <- forkIO (recvLoop state)+ sendThread <- forkIO (sendLoop state)+ ackThread <- forkIO (ackLoop state)+ retransmitLoop state++sendMessage :: Message -> Endpoint -> MessagingState -> IO ()+sendMessage message dstEndpoint (MessagingState transport store) = do+ srcEndpoint <- localEndpoint transport+ now <- getCurrentTime+ initialTimeout <- randomRIO (ackTimeout, ackTimeout * ackRandomFactor)+ {-putStrLn ("Queueing message " ++ (show message) ++ " for sending")-}+ let ctx = MessageContext { message = message+ , srcEndpoint = srcEndpoint+ , dstEndpoint = dstEndpoint }+ let messageState = MessageState { messageContext = ctx+ , insertionTime = now+ , replyTimeout = initialTimeout+ , retransmitCount = 0}+ atomically (queueMessage messageState (outgoingMessages store))++recvMessageMatching :: (MessageState -> Bool) -> MessagingState -> IO MessageContext+recvMessageMatching matchFilter (MessagingState _ store) = do+ msgState <- atomically (takeMessageRetryMatching matchFilter (incomingMessages store))+ let msgCtx = messageContext msgState+ let msgType = messageType (messageHeader (message msgCtx))+ when (msgType == CON) (atomically (queueMessage msgState (unackedMessages store)))+ return msgCtx+++cmpMessageCode :: MessageCode -> MessageCode -> Bool+cmpMessageCode (CodeRequest _) (CodeRequest _) = True+cmpMessageCode (CodeResponse _) (CodeResponse _) = True+cmpMessageCode CodeEmpty CodeEmpty = True+cmpMessageCode _ _ = False++recvRequest :: MessagingState -> IO MessageContext+recvRequest = recvMessageMatching (cmpMessageCode (CodeRequest GET) . messageCode . messageHeader . 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 }++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 }++allocateMessageId :: IO MessageId+allocateMessageId = randomIO+ ++sendResponse :: MessageContext -> Message -> MessagingState -> IO ()+sendResponse requestCtx response state@(MessagingState _ store) = do+ let origin = srcEndpoint requestCtx+ let reqToken = messageToken (message requestCtx)+ unackedMsg <- atomically (takeMessageByToken reqToken (unackedMessages store)) ++ msgId <- case unackedMsg of+ Nothing -> allocateMessageId+ _ -> return (messageId (messageHeader (message requestCtx)))+ + let outgoingMessage = case unackedMsg of+ Nothing -> setMessageId msgId response+ _ -> createAckResponse response++ sendMessage outgoingMessage origin state++sendRequest :: Message -> Endpoint -> MessagingState -> IO ()+sendRequest (Message (MessageHeader 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+ , messageToken = tkn+ , messageOptions = opts+ , messagePayload = payload }+ sendMessage msg msgdest state++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))+
+ src/Network/CoAP/Server.hs view
@@ -0,0 +1,86 @@+{-|+Module: Network.CoAP.Server+Description: CoAP server library+Maintainer: ulf.lilleengen@gmail.com+License: BSD3++The CoAP server API is intended to provide the minimal building block needed for creating CoAP servers. The API exposes CoAP requests and response types and handles all internal messaging details of the CoAP protocol.+-}+module Network.CoAP.Server+( Request(..)+, Method(..)+, Response(..)+, ResponseCode(..)+, Option(..)+, MediaType(..)+, createServer+, shutdownServer+, Server(..)+, RequestHandler+) where++import Network.CoAP.Messaging+import Network.CoAP.Types+import Control.Concurrent.Async+import Control.Concurrent.STM+import Network.Socket++-- | A request handler for a CoAP request. The request may be called by multiple threads+-- concurrently.+type RequestHandler = (Request, Endpoint) -> IO Response++-- | A CoAP server instance.+data Server = Server { runServer :: IO ()+ , msgThreadId :: Async () }+ ++-- | 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)+ return Server { runServer = requestLoop state handler+ , msgThreadId = msgThread }++-- | Shutdown a CoAP server.+shutdownServer :: Server -> IO ()+shutdownServer server = wait (msgThreadId server) -- TODO: Stop messaging thread.++createRequest :: MessageContext -> Request+createRequest reqCtx =+ let msg = message reqCtx+ (CodeRequest method) = messageCode (messageHeader msg)+ in Request { requestMethod = method+ , requestOptions = messageOptions msg + , requestPayload = messagePayload msg+ , requestReliable = messageType (messageHeader msg) == CON }++handleRequest :: MessageContext -> RequestHandler -> MessagingState -> IO ()+handleRequest requestCtx requestHandler state = do+ -- TODO: Add timeout+ let request = createRequest requestCtx+ {-putStrLn ("Received request: " ++ (show request))-}+ response <- requestHandler (request, srcEndpoint requestCtx)+ {-putStrLn ("Produced response: " ++ (show response))-}+ let responseMsg = createResponseMessage (message requestCtx) response+ sendResponse requestCtx responseMsg state++requestLoop :: MessagingState -> RequestHandler -> IO ()+requestLoop state requestHandler = do+ {-putStrLn "Waiting for incoming message"-}+ requestCtx <- recvRequest state+ _ <- async (handleRequest requestCtx requestHandler state)+ requestLoop state requestHandler++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 }+
+ src/Network/CoAP/Transport.hs view
@@ -0,0 +1,25 @@+{-|+Module: Network.CoAP.Transport+Description: CoAP transport library+Maintainer: ulf.lilleengen@gmail.com+License: BSD3++The CoAP transport API is intended to provide different transports available for CoAP. Currently only UDP transport is supported.+-}+module Network.CoAP.Transport+( createUDPTransport+, Endpoint(..)+, Transport(..)+) where++import Network.CoAP.Types+import qualified Network.Socket.ByteString as N+import Network.Socket hiding (send, sendTo, recv, recvFrom)+import Data.ByteString++-- | Create UDP transport instance for a given socket.+createUDPTransport :: Socket -> Transport+createUDPTransport sock =+ Transport { sendTo = N.sendTo sock+ , recvFrom = N.recvFrom sock 65535+ , localEndpoint = getSocketName sock }
+ src/Network/CoAP/Types.hs view
@@ -0,0 +1,126 @@+module Network.CoAP.Types where++import Data.ByteString.Lazy+import qualified Data.ByteString as BS+import Data.Word+import Data.Maybe+import Network.Socket++-- | A CoAP endpoint.+type Endpoint = SockAddr++-- | A CoAP transport instance capable of sending and receiving data+data Transport =+ Transport { sendTo :: BS.ByteString -> Endpoint -> IO Int+ , recvFrom :: IO (BS.ByteString, Endpoint)+ , localEndpoint :: IO Endpoint }++-- | Supported media types in CoAP RFC for ContentFormat+data MediaType = TextPlain+ | ApplicationLinkFormat+ | ApplicationXml+ | ApplicationOctetStream+ | ApplicationExi+ | ApplicationJson+ deriving (Show)++-- | CoAP Option ByteString+type OptionString = BS.ByteString++-- | CoAP Option types.+data Option = ContentFormat MediaType+ | ETag OptionString+ | LocationPath OptionString+ | LocationQuery OptionString+ | MaxAge Int+ | ProxyUri OptionString+ | ProxyScheme OptionString+ | UriHost OptionString+ | UriPath OptionString+ | UriPort Int+ | UriQuery OptionString+ | Accept Int+ | IfMatch OptionString+ | IfNoneMatch+ | Size1 Int+ deriving (Show)++-- | Request/response payload ByteString+type Payload = BS.ByteString++-- | Request Method+data Method = GET | POST | PUT | DELETE deriving (Show, Eq)++data MessageContext = MessageContext { message :: Message+ , srcEndpoint :: Endpoint+ , dstEndpoint :: Endpoint } deriving (Show)+-- | CoAP response.+data Response =+ Response { responseCode :: ResponseCode+ , responseOptions :: [Option]+ , responsePayload :: Maybe Payload } deriving (Show)++-- | CoAP request.+data Request =+ Request { requestMethod :: Method+ , requestOptions :: [Option]+ , requestPayload :: Maybe Payload+ , requestReliable :: Bool } deriving (Show)++-- | CoAP response codes.+data ResponseCode = Created+ | Deleted+ | Valid+ | Changed+ | Content+ | BadRequest+ | Unauthorized+ | BadOption+ | Forbidden+ | NotFound+ | MethodNotAllowed+ | NotAcceptable+ | PreconditionFailed+ | RequestEntityTooLarge+ | UnsupportedFormat+ | InternalServerError+ | NotImplemented+ | BadGateway+ | ServiceUnavailable+ | GatewayTimeout+ | ProxyingNotSupported+ deriving (Show, Eq)++-- | Valid CoAP message codes.+data MessageCode = CodeRequest Method+ | CodeResponse ResponseCode+ | CodeEmpty+ deriving (Show, Eq)++-- | 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.+type MessageId = Word16+-- | CoAP request token. This is unique for a request from a given endpoint.+type Token = BS.ByteString++-- | CoAP message record. Can be encoded and decoded to ByteString.+data Message = Message+ { messageHeader :: MessageHeader+ , messageToken :: Token+ , messageOptions :: [Option]+ , messagePayload :: Maybe Payload+ } deriving (Show)++
+ test/Spec.hs view
@@ -0,0 +1,136 @@+import Network.CoAP.Message+import qualified Network.CoAP.Server as S+import qualified Network.CoAP.Client as C+import Network.CoAP.Transport+import Network.Socket hiding (sendTo, recvFrom)+import Test.HUnit+import Data.ByteString.Char8 hiding (putStrLn)+import Test.QuickCheck+import Prelude hiding (null, length)+import Control.Concurrent.Chan+import Control.Concurrent+import Control.Monad+import System.Random+import System.Timeout+import Data.Maybe++data Cluster = Cluster Transport Transport++tests = TestList [TestLabel "testReliability" testReliability]++sendToChan prob chan msg endpoint = do+ v <- randomRIO(0, 1)+ when (v >= prob) (putStrLn ("Message to to " ++ (show endpoint) ++ " dropped!"))+ when (v < prob) (writeChan chan msg)+ return (length msg)++createUnstableTransport :: Double -> (Endpoint, Chan ByteString) -> (Endpoint, Chan ByteString) -> Transport+createUnstableTransport prob (localE, localC) (remoteE, remoteC) =+ Transport { sendTo = sendToChan prob remoteC+ , recvFrom = do+ d <- readChan localC+ return (d, remoteE)+ , localEndpoint = return localE }++testHandler :: S.RequestHandler+testHandler (req, _) = return (S.Response S.Created [] (Just (pack "Hello, Client")))++instance Arbitrary C.Request where+ arbitrary = do+ pload <- arbitrary+ return (C.Request { C.requestMethod = GET+ , C.requestOptions = []+ , C.requestPayload = pload+ , C.requestReliable = True })++testReliability =+ TestCase (do+ linkAProb <- randomRIO(0.7, 1)+ linkBProb <- randomRIO(0.7, 1)+ chanA <- newChan+ chanB <- newChan+ let endpointA = SockAddrUnix "A"+ let endpointB = SockAddrUnix "B"+ 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)+ client <- C.createClient transportB + reqs <- generate (vector 10)+ mapM_ (\req -> do+ response <- timeout 20000000 (C.doRequest client endpointA req)+ assertBool ("Timed out waiting for response on reliable request " ++ show req) (isJust response)+ let res = fromJust response+ putStrLn "Got response, checking"+ assertEqual "Bad response code" C.Created (C.responseCode res)+ assertEqual "Bad payload" "Hello, Client" (unpack (fromJust (C.responsePayload res)))+ ) reqs)++-- TODO, negative tests:+-- * Invalid token length+-- * Invalid code+-- * Unknown options+-- * Bad payload marker++instance Arbitrary MessageType where+ arbitrary = elements [CON, NON, ACK, RST]++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)))++instance Arbitrary MediaType where+ arbitrary = elements [TextPlain, ApplicationLinkFormat, ApplicationXml, ApplicationOctetStream, ApplicationExi, ApplicationJson]++instance Arbitrary Option where+ arbitrary = do+ mediaType <- arbitrary+ etag <- arbitrary+ loc <- arbitrary+ locquery <- arbitrary+ puri <- arbitrary+ pscheme <- arbitrary+ uriHost <- arbitrary+ uriPath <- arbitrary+ uriQuery <- arbitrary+ ifMatch <- arbitrary+ elements [ContentFormat mediaType, ETag etag, LocationPath loc, LocationQuery locquery, ProxyUri puri, ProxyScheme pscheme, UriHost uriHost, UriPath uriPath, UriQuery uriQuery, IfMatch ifMatch]++instance Arbitrary Message where+ arbitrary = do+ hdr <- arbitrary+ tkn <- arbitrary+ options <- arbitrary+ payload <- arbitrary+ return (Message { messageHeader = hdr+ , messageToken = tkn+ , messageOptions = [options]+ , messagePayload = payload })+++checkCodec msg =+ let encoded = encode msg+ decoded = decode encoded+ in (show msg) == (show decoded)++main :: IO ()+main = do+ putStrLn "Running HUnit tests"+ runTestTT tests+ putStrLn "Running QuickCheck tests"+ quickCheck checkCodec+ return ()+