diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/mediabus-demo-rtp-alaw-player/Main.hs b/mediabus-demo-rtp-alaw-player/Main.hs new file mode 100644 --- /dev/null +++ b/mediabus-demo-rtp-alaw-player/Main.hs @@ -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) diff --git a/mediabus-rtp.cabal b/mediabus-rtp.cabal new file mode 100644 --- /dev/null +++ b/mediabus-rtp.cabal @@ -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 diff --git a/specs/Data/MediaBus/Rtp/SourceSpec.hs b/specs/Data/MediaBus/Rtp/SourceSpec.hs new file mode 100644 --- /dev/null +++ b/specs/Data/MediaBus/Rtp/SourceSpec.hs @@ -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)))) diff --git a/specs/Spec.hs b/specs/Spec.hs new file mode 100644 --- /dev/null +++ b/specs/Spec.hs @@ -0,0 +1,1 @@ +{-# OPTIONS_GHC -F -pgmF hspec-discover #-} diff --git a/src/Data/MediaBus/Rtp.hs b/src/Data/MediaBus/Rtp.hs new file mode 100644 --- /dev/null +++ b/src/Data/MediaBus/Rtp.hs @@ -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 diff --git a/src/Data/MediaBus/Rtp/AlawSource.hs b/src/Data/MediaBus/Rtp/AlawSource.hs new file mode 100644 --- /dev/null +++ b/src/Data/MediaBus/Rtp/AlawSource.hs @@ -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 diff --git a/src/Data/MediaBus/Rtp/Packet.lhs b/src/Data/MediaBus/Rtp/Packet.lhs new file mode 100644 --- /dev/null +++ b/src/Data/MediaBus/Rtp/Packet.lhs @@ -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 diff --git a/src/Data/MediaBus/Rtp/Source.hs b/src/Data/MediaBus/Rtp/Source.hs new file mode 100644 --- /dev/null +++ b/src/Data/MediaBus/Rtp/Source.hs @@ -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)