mediabus-rtp (empty) → 0.2.0.0
raw patch · 10 files changed
+1163/−0 lines, 10 filesdep +QuickCheckdep +arraydep +asyncsetup-changed
Dependencies added: QuickCheck, array, async, base, binary, bytestring, cereal, conduit, conduit-combinators, conduit-extra, containers, data-default, deepseq, hspec, lens, lifted-async, mediabus, mediabus-rtp, monad-control, mtl, network, parallel, primitive, process, random, resourcet, singletons, spool, stm, streaming-commons, tagged, template-haskell, text, time, transformers, type-spec, vector
Files
- LICENSE +30/−0
- Setup.hs +2/−0
- mediabus-demo-rtp-alaw-player/Main.hs +41/−0
- mediabus-rtp.cabal +156/−0
- specs/Data/MediaBus/Rtp/SourceSpec.hs +219/−0
- specs/Spec.hs +1/−0
- src/Data/MediaBus/Rtp.hs +9/−0
- src/Data/MediaBus/Rtp/AlawSource.hs +39/−0
- src/Data/MediaBus/Rtp/Packet.lhs +549/−0
- src/Data/MediaBus/Rtp/Source.hs +117/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Sven Heyll, Lindenbaum GmbH (c) 2016++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 Sven Heyll 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
+ mediabus-demo-rtp-alaw-player/Main.hs view
@@ -0,0 +1,41 @@+module Main where++import System.Environment+import Data.MediaBus.AsyncConduit+import Conduit+import Data.MediaBus+import Data.MediaBus.Rtp++{- Send test data with:+#!/bin/bash++PORT=${1?port missing}+MY_IP=${2?host ip missing}+FNAME=${3:-28797-04.ogg}+FILE=$(realpath $(dirname ${0})/$FNAME)++gst-launch-1.0 uridecodebin uri=file://$FILE ! audioconvert ! audioresample ! audio/x-raw,format=S16LE,rate=8000,channels=1 ! alawenc ! rtppcmapay pt=8 mtu=172 min-ptime=10000000 max-ptime=200000000 ptime-multiple=5000000 ! udpsink host=$MY_IP port=$PORT+-}+maxFrames :: Int+maxFrames = 15000++main :: IO ()+main = do+ args <- getArgs+ case args of+ [] -> mainASync+ (_ : _) -> mainSync++mainASync :: IO ()+mainASync = runResourceT $+ withAsyncPolledSource 20+ (rtpAlaw16kHzS16Source 10000 "127.0.01" 5)+ (\(_, !src) -> runConduit (src .|+ exitAfterC maxFrames .|+ concealMissing blank .|+ streamDebugPlaybackSink))++mainSync :: IO ()+mainSync = runConduitRes (rtpAlaw16kHzS16Source 10000 "127.0.01" 20 .|+ exitAfterC maxFrames .|+ streamDebugPlaybackSink)
+ mediabus-rtp.cabal view
@@ -0,0 +1,156 @@+name: mediabus-rtp+version: 0.2.0.0+cabal-version: >=1.10+build-type: Simple+license: BSD3+license-file: LICENSE+copyright: 2016,2017 Sven Heyll, Lindenbaum GmbH+maintainer: sven.heyll@lindenbaum.eu+homepage: https://github.com/lindenbaum/mediabus-rtp+synopsis: Receive and Send RTP Packets+description:+ Please see README.md+category: Audio+author: Sven Heyll++source-repository head+ type: git+ location: https://github.com/lindenbaum/mediabus-rtp++library+ exposed-modules:+ Data.MediaBus.Rtp+ Data.MediaBus.Rtp.Packet+ Data.MediaBus.Rtp.Source+ Data.MediaBus.Rtp.AlawSource+ build-depends:+ QuickCheck >=2.9.2 && <2.10,+ array >=0.5.1.1 && <0.6,+ async >=2.1.1 && <2.2,+ base >=4.9 && <5,+ bytestring >=0.10.8.1 && <0.11,+ cereal >=0.5.4.0 && <0.6,+ conduit >=1.2.9 && <1.3,+ conduit-combinators >=1.1.0 && <1.2,+ conduit-extra >=1.1.15 && <1.2,+ containers >=0.5.7.1 && <0.6,+ data-default >=0.7.1.1 && <0.8,+ deepseq >=1.4.2.0 && <1.5,+ lens >=4.15.1 && <4.16,+ lifted-async >=0.9.1.1 && <0.10,+ mediabus >=0.2.0.2 && <0.3,+ monad-control >=1.0.1.0 && <1.1,+ mtl >=2.2.1 && <2.3,+ network >=2.6.3.1 && <2.7,+ parallel >=3.2.1.0 && <3.3,+ primitive >=0.6.1.0 && <0.7,+ process >=1.4.3.0 && <1.5,+ random ==1.1.*,+ resourcet >=1.1.9 && <1.2,+ spool ==0.1,+ stm >=2.4.4.1 && <2.5,+ streaming-commons >=0.1.17 && <0.2,+ tagged >=0.8.5 && <0.9,+ text >=1.2.2.1 && <1.3,+ time >=1.6.0.1 && <1.7,+ transformers >=0.5.2.0 && <0.6,+ vector >=0.11.0.0 && <0.12+ default-language: Haskell2010+ default-extensions: ApplicativeDo BangPatterns ConstraintKinds CPP+ DataKinds DefaultSignatures DeriveDataTypeable DeriveFoldable+ DeriveFunctor DeriveGeneric DeriveLift DeriveTraversable+ DuplicateRecordFields EmptyDataDecls EmptyCase FlexibleInstances+ FlexibleContexts FunctionalDependencies GADTs+ GeneralizedNewtypeDeriving KindSignatures LambdaCase+ MultiParamTypeClasses MultiWayIf NamedFieldPuns OverloadedStrings+ QuasiQuotes RecordWildCards RankNTypes ScopedTypeVariables+ StandaloneDeriving TemplateHaskell TupleSections TypeApplications+ TypeFamilies TypeInType TypeOperators TypeSynonymInstances+ UnicodeSyntax+ hs-source-dirs: src+ ghc-options: -O2 -Wall -funbox-strict-fields -fno-warn-unused-do-bind -fprint-explicit-kinds++executable mediabus-demo-rtp-alaw-player+ main-is: Main.hs+ build-depends:+ QuickCheck >=2.9.2 && <2.10,+ async >=2.1.1 && <2.2,+ base >=4.9 && <5,+ conduit >=1.2.9 && <1.3,+ conduit-combinators >=1.1.0 && <1.2,+ conduit-extra >=1.1.15 && <1.2,+ mediabus >=0.2.0.2 && <0.3,+ mediabus-rtp >=0.2.0.0 && <0.3,+ data-default >=0.7.1.1 && <0.8,+ lens >=4.15.1 && <4.16,+ lifted-async >=0.9.1.1 && <0.10,+ monad-control >=1.0.1.0 && <1.1,+ mtl >=2.2.1 && <2.3,+ parallel >=3.2.1.0 && <3.3,+ random ==1.1.*,+ stm >=2.4.4.1 && <2.5,+ tagged >=0.8.5 && <0.9,+ streaming-commons >=0.1.17 && <0.2,+ time >=1.6.0.1 && <1.7,+ vector >=0.11.0.0 && <0.12+ default-language: Haskell2010+ default-extensions: ApplicativeDo Arrows BangPatterns+ ConstraintKinds CPP DataKinds DefaultSignatures DeriveDataTypeable+ DeriveFoldable DeriveFunctor DeriveGeneric DeriveLift+ DeriveTraversable DuplicateRecordFields EmptyDataDecls EmptyCase+ FlexibleInstances FlexibleContexts FunctionalDependencies GADTs+ GeneralizedNewtypeDeriving KindSignatures LambdaCase+ MultiParamTypeClasses MultiWayIf NamedFieldPuns OverloadedStrings+ QuasiQuotes RankNTypes ScopedTypeVariables StandaloneDeriving+ TemplateHaskell TupleSections TypeApplications TypeFamilies+ TypeInType TypeOperators TypeSynonymInstances UnicodeSyntax+ hs-source-dirs: mediabus-demo-rtp-alaw-player+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -j +RTS -A256m -n2m -RTS -Wall -O2 -fno-warn-unused-binds -fno-warn-orphans -fno-warn-unused-do-bind -fno-warn-missing-signatures -fno-warn-type-defaults++test-suite tests+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ build-depends:+ QuickCheck >=2.9.2 && <2.10,+ array >=0.5.1.1 && <0.6,+ async >=2.1.1 && <2.2,+ base >=4.9 && <5,+ binary >=0.8.3.0 && <0.9,+ bytestring >=0.10.8.1 && <0.11,+ conduit >=1.2.9 && <1.3,+ conduit-combinators >=1.1.0 && <1.2,+ conduit-extra >=1.1.15 && <1.2,+ mediabus >=0.2.0.2 && <0.3,+ mediabus-rtp >=0.2.0.0 && <0.3,+ containers >=0.5.7.1 && <0.6,+ data-default >=0.7.1.1 && <0.8,+ deepseq >=1.4.2.0 && <1.5,+ hspec >=2.4.1 && <2.5,+ lens >=4.15.1 && <4.16,+ monad-control >=1.0.1.0 && <1.1,+ mtl >=2.2.1 && <2.3,+ singletons ==2.2.*,+ spool ==0.1,+ stm >=2.4.4.1 && <2.5,+ tagged >=0.8.5 && <0.9,+ template-haskell >=2.11.1.0 && <2.12,+ text >=1.2.2.1 && <1.3,+ time >=1.6.0.1 && <1.7,+ transformers >=0.5.2.0 && <0.6,+ type-spec >=0.3.0.1 && <0.4,+ vector >=0.11.0.0 && <0.12+ default-language: Haskell2010+ default-extensions: ApplicativeDo Arrows BangPatterns+ ConstraintKinds CPP DataKinds DefaultSignatures DeriveDataTypeable+ DeriveFoldable DeriveFunctor DeriveGeneric DeriveLift+ DeriveTraversable DuplicateRecordFields EmptyDataDecls EmptyCase+ FlexibleInstances FlexibleContexts FunctionalDependencies GADTs+ GeneralizedNewtypeDeriving KindSignatures LambdaCase+ MultiParamTypeClasses MultiWayIf NamedFieldPuns OverloadedStrings+ QuasiQuotes RankNTypes ScopedTypeVariables StandaloneDeriving+ TemplateHaskell TupleSections TypeApplications TypeFamilies+ TypeInType TypeOperators TypeSynonymInstances UnicodeSyntax+ hs-source-dirs: specs+ other-modules:+ Data.MediaBus.Rtp.SourceSpec+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -j +RTS -A256m -n2m -RTS -Wall -O0 -fno-warn-unused-binds -fno-warn-orphans -fno-warn-unused-do-bind -fno-warn-missing-signatures -fno-warn-type-defaults
+ specs/Data/MediaBus/Rtp/SourceSpec.hs view
@@ -0,0 +1,219 @@+module Data.MediaBus.Rtp.SourceSpec ( spec ) where++import Conduit+import Data.Conduit.List+import Test.Hspec+import Data.MediaBus.Rtp+import Data.MediaBus+import qualified Data.ByteString as B+import Data.Word+import Control.Lens+import Data.Proxy++spec :: Spec+spec = rtpSourceSpec >> rtpPayloadDemuxSpec++rtpSourceSpec :: Spec+rtpSourceSpec = describe "rtpSource" $ do+ let countStarts :: [Series a b] -> Int+ countStarts = foldr (\x n -> case x of+ Start _ -> n + 1+ Next _ -> n)+ 0+ runTestConduit inputs = runConduitPure (sourceList inputs .|+ annotateTypeCIn (Proxy :: Proxy (Stream Int Int Int () B.ByteString))+ rtpSource .|+ consume)++ it "yields 'Start' when only when the first payload packet arrives" $+ let inputs = [ MkStream (Start (MkFrameCtx 0 0 0 ())) ]+ in+ countStarts (_stream <$> runTestConduit inputs) `shouldBe`+ 0++ it "yields 'Start' when the first packet arrives" $+ let inputs = [ MkStream (Start (MkFrameCtx 0 0 0 ()))+ , mkTestRtpPacket 0 0 0+ ]+ in+ countStarts (_stream <$> runTestConduit inputs) `shouldBe`+ 1++ it "yields 'Start' when the ssrc changes" $+ let inputs = [ MkStream (Start (MkFrameCtx 8 9 10 ()))+ , mkTestRtpPacket ssrc0 0 0+ , mkTestRtpPacket ssrc0 0 0+ , mkTestRtpPacket ssrc1 0 0+ , mkTestRtpPacket ssrc1 0 0+ , mkTestRtpPacket ssrc0 0 0+ , mkTestRtpPacket ssrc1 0 0+ , mkTestRtpPacket ssrc1 0 0+ ]+ ssrc0 = 123+ ssrc1 = 345+ in+ countStarts (_stream <$> runTestConduit inputs) `shouldBe`+ 4++ it "yields 'Start' when the sequence numbers change too much" $+ let inputs = [ MkStream (Start (MkFrameCtx 0 0 0 ()))+ , mkTestRtpPacket 0 01 0+ , mkTestRtpPacket 0 02 0+ , mkTestRtpPacket 0 03 0+ , mkTestRtpPacket 0 24 0+ , mkTestRtpPacket 0 25 0+ , mkTestRtpPacket 0 06 0+ , mkTestRtpPacket 0 07 0+ ]+ in+ countStarts (_stream <$> runTestConduit inputs) `shouldBe`+ 3++ it "yields 'Start' when the sequence numbers change too much" $+ let inputs = [ MkStream (Start (MkFrameCtx 0 0 0 ()))+ , mkTestRtpPacket 0 01 0+ , mkTestRtpPacket 0 02 0+ , mkTestRtpPacket 0 03 0+ , mkTestRtpPacket 0 24 0+ , mkTestRtpPacket 0 25 0+ , mkTestRtpPacket 0 06 0+ , mkTestRtpPacket 0 07 0+ ]+ in+ countStarts (_stream <$> runTestConduit inputs) `shouldBe`+ 3++ it "yields no 'Start' when the sequence number wraps around" $+ let inputs = [ MkStream (Start (MkFrameCtx 0 0 0 ()))+ , mkTestRtpPacket 0 (negate 3) 0+ , mkTestRtpPacket 0 (negate 2) 0+ , mkTestRtpPacket 0 (negate 1) 0+ , mkTestRtpPacket 0 0 0+ , mkTestRtpPacket 0 1 0+ ]+ in+ countStarts (_stream <$> runTestConduit inputs) `shouldBe`+ 1++ it "yields no 'Start' when the timestamp wraps around" $+ let inputs = [ MkStream (Start (MkFrameCtx 0 0 0 ()))+ , mkTestRtpPacket 0 0 (negate 300)+ , mkTestRtpPacket 0 0 (negate 200)+ , mkTestRtpPacket 0 0 (negate 100)+ , mkTestRtpPacket 0 0 0+ , mkTestRtpPacket 0 0 100+ ]+ in+ countStarts (_stream <$> runTestConduit inputs) `shouldBe`+ 1++ it "can handle broken packets without crashing" $+ let inputs = [ MkStream (Start (MkFrameCtx 0 0 0 ()))+ , mkTestRtpPacket 0 0 777+ , mkTestRtpPacket 0 0 777+ , mkBrokenTestRtpPacket+ , mkBrokenTestRtpPacket+ , mkTestRtpPacket 0 0 777+ , mkTestRtpPacket 0 0 777+ , mkBrokenTestRtpPacket+ , mkTestRtpPacket 0 0 777+ , mkTestRtpPacket 0 0 777+ , mkTestRtpPacket 0 0 777+ ]+ in+ length (runTestConduit inputs) `shouldBe`+ 8++rtpPayloadDemuxSpec :: Spec+rtpPayloadDemuxSpec = describe "rtpPayloadDemux" $ do+ let runTestConduit inputs payloadHandlers fallback =+ runConduitPure (sourceList inputs .|+ rtpSource .|+ rtpPayloadDemux payloadHandlers fallback .|+ consume)++ it "always yields the fallback element if the payload table contains no handler" $+ let inputs = [ MkStream (Start (MkFrameCtx 0 0 0 ()))+ , mkTestRtpPacket 0 0 0+ ]+ outs = preview payload <$> runTestConduit inputs [] ()+ in+ outs `shouldBe` [ Nothing, Just () ]+ it "always yields the fallback element if the payload table contains no handler for the payload type" $+ let inputs = MkStream (Start (MkFrameCtx 0 0 0 ())) :+ [ mkTestRtpPacketWithPayload 0+ 0+ 0+ (mkTestPayload (MkRtpPayloadType p))+ | p <- [0 .. 128] ]+ fallback :: SampleBuffer (S16 8000)+ fallback = mempty+ outs = preview payload <$> runTestConduit inputs+ [ ( MkRtpPayloadType 129+ , transcode .+ alawPayloadHandler+ )+ ]+ fallback+ in+ outs `shouldBe` Nothing :+ [ Just fallback+ | _ <- [0 .. 128 :: Word8] ]+ it "invokes the first matching payload handler" $+ let inputs = MkStream (Start (MkFrameCtx 0 0 0 ())) :+ [ mkTestRtpPacketWithPayload 0 0 0 (mkTestPayload 8)+ , mkTestRtpPacketWithPayload 0 0 0 (mkTestPayload 0)+ ]+ outputs = preview payload <$> runTestConduit inputs+ [ ( 8+ , payload .~+ "first 8 handler"+ )+ , ( 8+ , payload .~+ "second 8 handler"+ )+ , ( 0+ , payload .~+ "first 0 handler"+ )+ , ( 0+ , payload .~+ "second 0 handler"+ )+ ]+ "bad"+ in+ outputs `shouldBe`+ [ Nothing, Just "first 8 handler", Just "first 0 handler" ]++mkBrokenTestRtpPacket :: Stream Int Int Int () B.ByteString+mkBrokenTestRtpPacket = MkStream (Next (MkFrame 0 0 (B.pack [ 0, 0, 0 ])))++mkTestPayload :: RtpPayloadType -> RtpPayload+mkTestPayload pt = MkRtpPayload pt (sampleBufferFromList [ 0, 0, 0 ])++mkTestRtpPacket :: RtpSsrc+ -> RtpSeqNum+ -> RtpTimestamp+ -> Stream Int Int Int () B.ByteString+mkTestRtpPacket ssrc sn ts =+ mkTestRtpPacketWithPayload ssrc sn ts (mkTestPayload 0)++mkTestRtpPacketWithPayload :: RtpSsrc+ -> RtpSeqNum+ -> RtpTimestamp+ -> RtpPayload+ -> Stream Int Int Int () B.ByteString+mkTestRtpPacketWithPayload ssrc sn ts p =+ MkStream (Next (MkFrame 0+ 0+ (serialize (MkRtpPacket (MkRtpHeader 2+ False+ False+ sn+ ts+ ssrc+ []+ Nothing)+ p))))
+ specs/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}
+ src/Data/MediaBus/Rtp.hs view
@@ -0,0 +1,9 @@+{-# OPTIONS -Wno-unused-top-binds #-}++module Data.MediaBus.Rtp+ ( module X+ ) where++import Data.MediaBus.Rtp.Source as X+import Data.MediaBus.Rtp.AlawSource as X+import Data.MediaBus.Rtp.Packet as X
+ src/Data/MediaBus/Rtp/AlawSource.hs view
@@ -0,0 +1,39 @@+module Data.MediaBus.Rtp.AlawSource ( rtpAlaw16kHzS16Source ) where++import Conduit+import Data.MediaBus.Audio.Raw+import Data.MediaBus.Audio.Resample+import Data.MediaBus.Segment+import Data.MediaBus.Clock+import Data.MediaBus.Reorder+import Data.MediaBus.Rtp.Source+import Data.MediaBus.Rtp.Packet+import Data.MediaBus.Sample+import Data.MediaBus.Stream+import Data.MediaBus.StreamSegment+import Data.MediaBus.SourceId+import Data.MediaBus.Conduit+import Data.MediaBus.Ticks+import Data.MediaBus.Transcoder+import Data.MediaBus.Transport.Udp+import Data.Streaming.Network ( HostPreference )+import Data.Word+import Data.Proxy+import qualified Data.ByteString as B+import Network.Socket ( SockAddr )++rtpAlaw16kHzS16Source :: MonadResource m+ => Int+ -> HostPreference+ -> Int+ -> Source m (Stream RtpSsrc RtpSeqNum (Ticks 16000 Word64) () (Segment (640 :@ 16000) (SampleBuffer (S16 16000))))+rtpAlaw16kHzS16Source !udpListenPort !udpListenIP !reorderBufferSize =+ annotateTypeSource (Proxy :: Proxy (Stream (SourceId (Maybe SockAddr)) RtpSeqNum (ClockTimeDiff UtcClock) () B.ByteString))+ (udpDatagramSource useUtcClock udpListenPort udpListenIP) .|+ rtpSource .|+ rtpPayloadDemux [ (8, alawPayloadHandler) ] mempty .|+ transcodeStreamC' .|+ resample8to16kHz' (MkS16 0 :: S16 8000) .|+ convertTicksC' at8kHzU32 at16kHzU64 .|+ reorderFramesBySeqNumC reorderBufferSize .|+ segmentC
+ src/Data/MediaBus/Rtp/Packet.lhs view
@@ -0,0 +1,549 @@+Serialize and deserialize RTP packets.+TODO: Add RTCP support++> module Data.MediaBus.Rtp.Packet+> ( RtpPacket(..), RtpHeader(..), HeaderExtension(..)+> , type RtpSeqNum, RtpSsrc(..), RtpTimestamp(..)+> , RtpPayloadType(..), rtpPayloadTypeValue+> , RtpPayload(..), rtpPayloadType, rtpPayload+> , deserialize, serialize)+> where++> import qualified Data.ByteString as B+> import Control.Monad+> import Data.Default+> import Data.Maybe+> import Data.Serialize.Get+> import Data.Serialize.Put+> import Text.Printf+> import Data.Word+> import Data.Bits+> import Data.MediaBus.Monotone+> import Data.MediaBus.Sequence+> import Control.Lens+> import Data.MediaBus.Sample+> import GHC.Generics ( Generic )+> import Control.DeepSeq+> import System.Random++The relevant output will be contained in the 'RtpPacket' and 'RtpHeader' data+types.++> data RtpPacket =+> MkRtpPacket { header :: !RtpHeader+> , body :: !RtpPayload }+> deriving (Eq, Generic)++The rtp header defines the logical source(s), the sequence number and the+timestamp of the respecitive 'RtpPayload'.++> data RtpHeader =+> MkRtpHeader { version :: !Word8+> , hasPadding :: !Bool+> , hasMarker :: !Bool+> , sequenceNumber :: !RtpSeqNum+> , headerTimestamp :: !RtpTimestamp+> , ssrc :: !RtpSsrc+> , csrcs :: ![RtpSsrc]+> , headerExtension :: !(Maybe HeaderExtension)}+> deriving (Eq, Generic)++An SSRC is basically just a 'Word32'.++> newtype RtpSsrc = MkRtpSsrc { rtpSsrc :: Word32 }+> deriving (Eq, Ord, Num, Bits, Default, Generic, Random)++> instance NFData RtpSsrc++> instance Show RtpSsrc where+> show (MkRtpSsrc w) = printf "ssrc:%10d" w++A timestamp is basically just a 'Word32', too.++> newtype RtpTimestamp = MkRtpTimestamp { _rtpTimestamp :: Word32 }+> deriving (Eq, Num, Bits, Default, Generic)++> instance NFData RtpTimestamp++> instance Show RtpTimestamp where+> show (MkRtpTimestamp w) = printf "ts:%10d" w++The 'Ord' instance of the 'RtpTimestamp' should handle the wrap-around+correctly, therefore we use the 'IsMonotonic' method 'succeeds'.++> instance Ord RtpTimestamp where+> (MkRtpTimestamp l) `compare` (MkRtpTimestamp r) =+> if l == r then EQ else if l `succeeds` r then GT else LT++SeqNum numbers are special because they wrap-around.++To meaningfully compare them, this must be taken into account.+E.g. when @x1 = 65535@ is the current sequence number and the next+packet has @x2 = 0@ then in this context @x2 > x1@.++A type alias to 'SeqNum' will ensure that 'Ord' respects the wrap-around,+such that for example @0 <= 1@ and @65535 <= 0@.++> type RtpSeqNum = SeqNum Word16++The 'HeaderExtension' is a profile specific, variable length, data block+following the fixed size RTP header:++> data HeaderExtension =+> MkHeaderExtension { headerExtensionField :: !Word16+> , headerExtensionBody :: ![Word32] }+> deriving (Read,Eq,Show, Generic)++> instance NFData HeaderExtension++A payload type is basically just a 'Word8':++> newtype RtpPayloadType = MkRtpPayloadType { _rtpPayloadTypeValue :: Word8 }+> deriving (Eq, Num, Bits, Default, Generic)++> instance NFData RtpPayloadType++> instance Show RtpPayloadType where+> show (MkRtpPayloadType w) = printf "pt:%3d" w++The payload contains the actual media data, i.e. the raw payload bytes together+with the 'RtpPayloadType'.++> data RtpPayload = MkRtpPayload { _rtpPayloadType :: RtpPayloadType+> , _rtpPayload :: SampleBuffer Word8+> }+> deriving (Eq, Generic)++> instance NFData RtpPayload++> makeLenses ''RtpPayloadType+> makeLenses ''RtpPayload++Deserialize a complete RTP datagram:++> deserialize :: B.ByteString -> Either String RtpPacket+> deserialize = runGet getPacket++Below are only internal functions.++This function will parse an 'Rtp' packet from a 'ByteString':++> getPacket :: Get RtpPacket+> getPacket = do++First read the header:++> (pt, h) <- getPayloadTypeAndHeader++Then read the remaining bytes:++> remainingLen <- remaining+> remainingBytes <- getBytes remainingLen++And then adjust for padding:++> let bodyBytes = if hasPadding h+> then adjustPadding remainingBytes+> else remainingBytes+> body = MkRtpPayload pt (sampleBufferFromByteString bodyBytes)++Wrap everything up and return it:++> return (MkRtpPacket h body)++Ok now to adjust for padding:++> adjustPadding :: B.ByteString -> B.ByteString+> adjustPadding padded =+> let paddingLen = B.last padded+> unpaddedLen = B.length padded - fromIntegral paddingLen+> in B.take unpaddedLen padded++This function will parse an 'RtpHeader':++> getPayloadTypeAndHeader :: Get (RtpPayloadType, RtpHeader)+> getPayloadTypeAndHeader = do++The values are in network byte order, i.e. big-endian.++> b1 <- getWord8++From https://www.ietf.org/rfc/rfc3550.txt, section 5:++ The RTP header has the following format:+ 0 1 2 3+ 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-++ |V=2|P|X| CC |M| PT | sequence number |+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-++ | timestamp |+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-++ | synchronization source (SSRC) identifier |+ +=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=++ | contributing source (CSRC) identifiers |+ | .... |+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-++++ The first twelve octets are present in every RTP packet, while the+ list of CSRC identifiers is present only when inserted by a mixer.+ The fields have the following meaning:++ version (V): 2 bits+ This field identifies the version of RTP. The version defined by+ this specification is two (2). (The value 1 is used by the first+ draft version of RTP and the value 0 is used by the protocol+ initially implemented in the "vat" audio tool.)++> let version' = fromIntegral ((b1 `shiftR` 6) .&. 3)++ padding (P): 1 bit+ If the padding bit is set, the packet contains one or more+ additional padding octets at the end which are not part of the+ payload. The last octet of the padding contains a count of how+ many padding octets should be ignored, including itself. Padding+ may be needed by some encryption algorithms with fixed block sizes+ or for carrying several RTP packets in a lower-layer protocol data+ unit.++> let hasPadding' = testBit b1 5++ extension (X): 1 bit+ If the extension bit is set, the fixed header MUST be followed by+ exactly one header extension, with a format defined in Section+ 5.3.1.++> let hasExtension = testBit b1 4++ CSRC count (CC): 4 bits+ The CSRC count contains the number of CSRC identifiers that follow+ the fixed header.++> let csrcCount = b1 .&. 0xf++ marker (M): 1 bit+ The interpretation of the marker is defined by a profile. It is+ intended to allow significant events such as frame boundaries to+ be marked in the packet stream. A profile MAY define additional+ marker bits or specify that there is no marker bit by changing the+ number of bits in the payload type field (see Section 5.3).++> b2 <- getWord8+> let hasMarker' = testBit b2 7++ payload type (PT): 7 bits+ This field identifies the format of the RTP payload and determines+ its interpretation by the application. A profile MAY specify a+ default static mapping of payload type codes to payload formats.+ Additional payload type codes MAY be defined dynamically through+ non-RTP means (see Section 3). A set of default mappings for+ audio and video is specified in the companion RFC 3551 [1]. An+ RTP source MAY change the payload type during a session, but this+ field SHOULD NOT be used for multiplexing separate media streams+ (see Section 5.2).++ A receiver MUST ignore packets with payload types that it does not+ understand.++> let payloadType' = fromIntegral (b2 .&. 0x7f)++ sequence number: 16 bits+ The sequence number increments by one for each RTP data packet+ sent, and may be used by the receiver to detect packet loss and to+ restore packet sequence. The initial value of the sequence number+ SHOULD be random (unpredictable) to make known-plaintext attacks+ on encryption more difficult, even if the source itself does not+ encrypt according to the method in Section 9.1, because the+ packets may flow through a translator that does. Techniques for+ choosing unpredictable numbers are discussed in [17].++> sequenceNumber' <- getWord16be++ timestamp: 32 bits+ The timestamp reflects the sampling instant of the first octet in+ the RTP data packet. The sampling instant MUST be derived from a+ clock that increments monotonically and linearly in time to allow+ synchronization and jitter calculations (see Section 6.4.1). The+ resolution of the clock MUST be sufficient for the desired+ synchronization accuracy and for measuring packet arrival jitter+ (one tick per video frame is typically not sufficient). The clock+ frequency is dependent on the format of data carried as payload+ and is specified statically in the profile or payload format+ specification that defines the format, or MAY be specified+ dynamically for payload formats defined through non-RTP means. If+ RTP packets are generated periodically, the nominal sampling+ instant as determined from the sampling clock is to be used, not a+ reading of the system clock. As an example, for fixed-rate audio+ the timestamp clock would likely increment by one for each+ sampling period. If an audio application reads blocks covering+ 160 sampling periods from the input device, the timestamp would be+ increased by 160 for each such block, regardless of whether the+ block is transmitted in a packet or dropped as silent.++ The initial value of the timestamp SHOULD be random, as for the+ sequence number. Several consecutive RTP packets will have equal+ timestamps if they are (logically) generated at once, e.g., belong+ to the same video frame. Consecutive RTP packets MAY contain+ timestamps that are not monotonic if the data is not transmitted+ in the order it was sampled, as in the case of MPEG interpolated+ video frames. (The sequence numbers of the packets as transmitted+ will still be monotonic.)++ RTP timestamps from different media streams may advance at+ different rates and usually have independent, random offsets.+ Therefore, although these timestamps are sufficient to reconstruct+ the timing of a single stream, directly comparing RTP timestamps+ from different media is not effective for synchronization.+ Instead, for each medium the RTP timestamp is related to the+ sampling instant by pairing it with a timestamp from a reference+ clock (wallclock) that represents the time when the data+ corresponding to the RTP timestamp was sampled. The reference+ clock is shared by all media to be synchronized. The timestamp+ pairs are not transmitted in every data packet, but at a lower+ rate in RTCP SR packets as described in Section 6.4.++ The sampling instant is chosen as the point of reference for the+ RTP timestamp because it is known to the transmitting endpoint and+ has a common definition for all media, independent of encoding+ delays or other processing. The purpose is to allow synchronized+ presentation of all media sampled at the same time.++ Applications transmitting stored data rather than data sampled in+ real time typically use a virtual presentation timeline derived+ from wallclock time to determine when the next frame or other unit+ of each medium in the stored data should be presented. In this+ case, the RTP timestamp would reflect the presentation time for+ each unit. That is, the RTP timestamp for each unit would be+ related to the wallclock time at which the unit becomes current on+ the virtual presentation timeline. Actual presentation occurs+ some time later as determined by the receiver.++ An example describing live audio narration of prerecorded video+ illustrates the significance of choosing the sampling instant as+ the reference point. In this scenario, the video would be+ presented locally for the narrator to view and would be+ simultaneously transmitted using RTP. The "sampling instant" of a+ video frame transmitted in RTP would be established by referencing+ its timestamp to the wallclock time when that video frame was+ presented to the narrator. The sampling instant for the audio RTP+ packets containing the narrator's speech would be established by+ referencing the same wallclock time when the audio was sampled.+ The audio and video may even be transmitted by different hosts if+ the reference clocks on the two hosts are synchronized by some+ means such as NTP. A receiver can then synchronize presentation+ of the audio and video packets by relating their RTP timestamps+ using the timestamp pairs in RTCP SR packets.++> timestamp' <- getWord32be++ SSRC: 32 bits+ The SSRC field identifies the synchronization source. This+ identifier SHOULD be chosen randomly, with the intent that no two+ synchronization sources within the same RTP session will have the+ same SSRC identifier. An example algorithm for generating a+ random identifier is presented in Appendix A.6. Although the+ probability of multiple sources choosing the same identifier is+ low, all RTP implementations must be prepared to detect and+ resolve collisions. Section 8 describes the probability of+ collision along with a mechanism for resolving collisions and+ detecting RTP-level forwarding loops based on the uniqueness of+ the SSRC identifier. If a source changes its source transport+ address, it must also choose a new SSRC identifier to avoid being+ interpreted as a looped source (see Section 8.2).++> ssrc' <- getWord32be++ CSRC list: 0 to 15 items, 32 bits each+ The CSRC list identifies the contributing sources for the payload+ contained in this packet. The number of identifiers is given by+ the CC field. If there are more than 15 contributing sources,+ only 15 can be identified. CSRC identifiers are inserted by+ mixers (see Section 7.1), using the SSRC identifiers of+ contributing sources. For example, for audio packets the SSRC+ identifiers of all sources that were mixed together to create a+ packet are listed, allowing correct talker indication at the+ receiver.++> csrcs' <- sequence (replicate (fromIntegral csrcCount) getWord32be)++If a the extension flag is set, we must parse an optional header extension:++> extension <- if hasExtension+> then Just <$> getHeaderExtension+> else return Nothing++> return ( payloadType'+> , MkRtpHeader+> { version = version'+> , hasPadding = hasPadding'+> , hasMarker = hasMarker'+> , sequenceNumber = MkSeqNum sequenceNumber'+> , headerTimestamp = MkRtpTimestamp timestamp'+> , ssrc = MkRtpSsrc ssrc'+> , csrcs = MkRtpSsrc <$> csrcs'+> , headerExtension = extension })+++The RFC defines how RTP header extensions must be provided.++Quoting the RFC again:++ 0 1 2 3+ 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-++ | defined by profile | length |+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-++ | header extension |+ | .... |++ If the X bit in the RTP header is one, a variable-length header+ extension MUST be appended to the RTP header, following the CSRC list+ if present. The header extension contains a 16-bit length field that+ counts the number of 32-bit words in the extension, excluding the+ four-octet extension header (therefore zero is a valid length). Only+ a single extension can be appended to the RTP data header.++> getHeaderExtension :: Get HeaderExtension+> getHeaderExtension = do+> field <- getWord16be+> len <- getWord16be+> body <- sequence (replicate (fromIntegral len) getWord32be)+> return (MkHeaderExtension field body)++NOTE: To test this, you can use gstreamer, e.g. with this command line:++@@@+gst-launch-1.0 autoaudiosrc is-live=true ! audioconvert ! audioresample ! alawenc ! rtppcmapay pt=8 mtu=172 ssrc=123345 ! udpsink host=someip port=theport+@@@++Here are the type class instances:++> instance Show RtpPacket where+> show (MkRtpPacket hdr bd) =+> "(RTP: " ++ show hdr ++ ", " ++ show bd ++ ")"++> instance Show RtpHeader where+> show (MkRtpHeader _ _ m s ts ssrc _csrcs hes) =+> printf "%s/%s/seq:%05d/m:%d/e:%s"+> (show ssrc)+> (show ts)+> (_fromSeqNum s)+> (if m then 0 else 1::Int)+> (maybe "0" (("|" ++) . show) hes)++> instance Show RtpPayload where+> show (MkRtpPayload pt bd) =+> printf "%s/%s" (show pt) (show bd)++Serialization is straight forward the opposite of deserialization.++> serialize :: RtpPacket -> B.ByteString+> serialize pkg = runPut (putPacket pkg)++> putPacket :: RtpPacket -> Put+> putPacket (MkRtpPacket h b) = do++First write the header then the body.++> putPayloadTypeAndHeader (_rtpPayloadType b) h+> putByteString (byteStringFromSampleBuffer (_rtpPayload b))++Calculate number of bytes required for padding.++> let paddingLen = fromIntegral+> ((64 - ((sampleCount (_rtpPayload b)) `rem` 64)) `rem` 64)++The 'Header' field 'hasPadding', which is an input to this function,+is interpreted to indicate if padding is /allowed/.++> when (hasPadding h && paddingLen > 0) $ do++Generate the padding. The last byte of the padding must contain the padding+length. The contents of the (other) padding bytes are ignored, so we can use+a little Haskell trick to generate a list of bytes with the last containing the+number of bytes in that list.++> let padding = [1 .. paddingLen]+> mapM_ putWord8 padding++Writing out the header:++> putPayloadTypeAndHeader :: RtpPayloadType -> RtpHeader -> Put+> putPayloadTypeAndHeader payloadType' MkRtpHeader{..} = do++To repeat the RTP header structure:++ 0 1 2 3+ 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-++ |V=2|P|X| CC |M| PT | sequence number |+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-++ | timestamp |+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-++ | synchronization source (SSRC) identifier |+ +=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=++ | contributing source (CSRC) identifiers |+ | .... |+ +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+++The first byte is the hardest:++> putWord8++Set the extension header flag++> (setBitTo 4 (isJust headerExtension)++Set the padding flag:++> (setBitTo 5 hasPadding++Set the version:++> (shiftL version 6 .|.++Write the number of CSRCs:++> (fromIntegral (length csrcs) .&. 0xf))))++The second byte contains the marker and the payload type:++> putWord8+> (setBitTo 7 hasMarker+> ((payloadType' ^. rtpPayloadTypeValue) .&. 0x7f))++The sequence number and timestamp:++> putWord16be (_fromSeqNum sequenceNumber)+> putWord32be (_rtpTimestamp headerTimestamp)++The SSRC:++> putWord32be (rtpSsrc ssrc)++The maximum of 16 csrcs:++> mapM_ putWord32be (rtpSsrc <$> (take 16 csrcs))++And last but not least the header extensions:++> mapM_ putHeaderExtension headerExtension++Serialize the 'HeaderExtension':++> putHeaderExtension :: HeaderExtension -> Put+> putHeaderExtension MkHeaderExtension{..} = do+> putWord16be headerExtensionField++Write the number of 'Word32's that make up the header extension body. Limit the+length to 0xffff since the length field is only 16 bits wide.++> let bodyLen = min 0xffff (length headerExtensionBody)+> putWord16be (fromIntegral bodyLen)+> mapM_ putWord32be (take bodyLen headerExtensionBody)++A litte binary helper:++> setBitTo :: Bits a => Int -> Bool -> a -> a+> setBitTo i cond a =+> (if cond then setBit else clearBit) a i
+ src/Data/MediaBus/Rtp/Source.hs view
@@ -0,0 +1,117 @@+{-# OPTIONS -Wno-unused-top-binds #-}++module Data.MediaBus.Rtp.Source+ ( type RtpStream+ , rtpSource+ , rtpPayloadDemux+ , type RtpPayloadHandler+ , alawPayloadHandler+ ) where++import Conduit+import Control.Lens+import qualified Data.ByteString as B+import Data.MediaBus.Audio.Alaw+import Data.MediaBus.Ticks+import Data.MediaBus.Sample+import Data.MediaBus.Stream+import Data.MediaBus.Series+import Data.MediaBus.Rtp.Packet+import Control.Monad+import Data.Default+import Text.Printf+import Debug.Trace+import Data.Coerce+import qualified Data.List+import Data.Maybe++type RtpStream p = Stream RtpSsrc RtpSeqNum RtpTimestamp p RtpPayload++data RRState ctx = MkRRState { _currCtx :: ctx+ , _isFirstPacket :: Bool+ }+ deriving (Show)++makeLenses ''RRState++rtpSource :: (Default i, Monad m, Default p, Show p)+ => Conduit (Stream i s t p B.ByteString) m (RtpStream p)+rtpSource = evalStateC (MkRRState (MkFrameCtx def def def def) True) $+ awaitForever processFrames+ where+ processFrames (MkStream (Start _)) =+ return ()+ processFrames (MkStream (Next (MkFrame _ _ !contentIn))) =+ case deserialize contentIn of+ Left rtpError -> traceRtpError rtpError+ Right rtpPacket -> do+ let rtpHeader = header rtpPacket+ res <- updateState rtpHeader+ when (res == FrameCtxChanged) yieldStreamStart+ yieldStreamNext (body rtpPacket)++ traceRtpError e = do+ ctx <- use currCtx+ traceM (printf "RTP-ERROR:%s Ctx: %s\n" e (show ctx))++ updateState rtpHeader = do+ oldCtx <- currCtx <<%=+ ((frameCtxSeqNumRef .~ sequenceNumber rtpHeader)+ . (frameCtxTimestampRef .~+ headerTimestamp rtpHeader))+ wasFirstPacket <- isFirstPacket <<.= False+ if oldCtx ^. frameCtxSourceId /= ssrc rtpHeader+ then do+ currCtx . frameCtxSourceId .= ssrc rtpHeader+ return FrameCtxChanged+ else if sequenceNumbersDifferTooMuch (oldCtx ^. frameCtxSeqNumRef)+ (sequenceNumber rtpHeader) ||+ timestampsDifferTooMuch (oldCtx ^. frameCtxTimestampRef)+ (headerTimestamp rtpHeader) ||+ wasFirstPacket+ then return FrameCtxChanged+ else return FrameCtxNotChanged+ where+ sequenceNumbersDifferTooMuch oldSN currSN =+ let d = if currSN >= oldSN then currSN - oldSN else oldSN - currSN -- TODO use LocalOrd??+ sequenceNumberMaxDelta =+ 10+ in+ d >= sequenceNumberMaxDelta+ timestampsDifferTooMuch oldTS currTS =+ let d = if currTS >= oldTS then currTS - oldTS else oldTS - currTS+ timestampMaxDelta = 2000 -- TODO extract+ in+ d >= timestampMaxDelta+ yieldStreamStart = use currCtx >>= yieldStartFrameCtx+ yieldStreamNext p = do+ ts <- use (currCtx . frameCtxTimestampRef)+ sn <- use (currCtx . frameCtxSeqNumRef)+ yield (MkStream (Next (MkFrame ts sn p)))++data RRSourceChange = FrameCtxChanged | FrameCtxNotChanged+ deriving (Eq)++rtpPayloadDemux :: (Integral t, Monad m)+ => [(RtpPayloadType, RtpPayloadHandler (Ticks r t) c)]+ -> c+ -> Conduit (RtpStream p) m (Stream RtpSsrc RtpSeqNum (Ticks r t) p c)+rtpPayloadDemux payloadTable fallbackContent =+ mapC (timestamp %~ (MkTicks . fromIntegral . _rtpTimestamp)) .|+ awaitForever go+ where+ setFallbackContent = payload .~ fallbackContent+ go (MkStream (Next !frm)) =+ let pt = frm ^. framePayload . rtpPayloadType+ mHandler = Data.List.lookup pt payloadTable+ !frm' = fromMaybe setFallbackContent mHandler frm+ in+ yieldNextFrame frm'+ go (MkStream (Start !frmCtx)) =+ yieldStartFrameCtx frmCtx++type RtpPayloadHandler t c = Frame RtpSeqNum t RtpPayload+ -> Frame RtpSeqNum t c++alawPayloadHandler :: RtpPayloadHandler t (SampleBuffer ALaw)+alawPayloadHandler = payload %~ (coerce . _rtpPayload)