diff --git a/mediabus-demo-rtp-alaw-player/Main.hs b/mediabus-demo-rtp-alaw-player/Main.hs --- a/mediabus-demo-rtp-alaw-player/Main.hs +++ b/mediabus-demo-rtp-alaw-player/Main.hs @@ -1,9 +1,9 @@ module Main where import System.Environment -import Data.MediaBus.AsyncConduit import Conduit import Data.MediaBus +import Control.Monad.Logger import Data.MediaBus.Rtp {- Send test data with: @@ -27,15 +27,15 @@ (_ : _) -> mainSync mainASync :: IO () -mainASync = runResourceT $ +mainASync = runStdoutLoggingT $ runResourceT $ withAsyncPolledSource 20 (rtpAlaw16kHzS16Source 10000 "127.0.01" 5) (\(_, !src) -> runConduit (src .| exitAfterC maxFrames .| concealMissing blank .| - streamDebugPlaybackSink)) + debugAudioPlaybackSink)) mainSync :: IO () -mainSync = runConduitRes (rtpAlaw16kHzS16Source 10000 "127.0.01" 20 .| - exitAfterC maxFrames .| - streamDebugPlaybackSink) +mainSync = runStdoutLoggingT $ runConduitRes (rtpAlaw16kHzS16Source 10000 "127.0.01" 20 .| + exitAfterC maxFrames .| + debugAudioPlaybackSink) diff --git a/mediabus-rtp.cabal b/mediabus-rtp.cabal --- a/mediabus-rtp.cabal +++ b/mediabus-rtp.cabal @@ -1,5 +1,5 @@ name: mediabus-rtp -version: 0.2.0.0 +version: 0.3.2.0 cabal-version: >=1.10 build-type: Simple license: BSD3 @@ -38,8 +38,9 @@ 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, + mediabus >=0.3.2.0 && <0.4, monad-control >=1.0.1.0 && <1.1, + monad-logger >=0.3.20.2 && <0.4, mtl >=2.2.1 && <2.3, network >=2.6.3.1 && <2.7, parallel >=3.2.1.0 && <3.3, @@ -79,12 +80,13 @@ 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, + mediabus >=0.3.2.0 && <0.4, + mediabus-rtp >=0.3.2.0 && <0.4, 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, + monad-logger >=0.3.20.2 && <0.4, mtl >=2.2.1 && <2.3, parallel >=3.2.1.0 && <3.3, random ==1.1.*, @@ -120,14 +122,15 @@ 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, + mediabus >=0.3.2.0 && <0.4, + mediabus-rtp >=0.3.2.0 && <0.4, 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, + monad-logger >=0.3.20.2 && <0.4, mtl >=2.2.1 && <2.3, singletons ==2.2.*, spool ==0.1, diff --git a/specs/Data/MediaBus/Rtp/SourceSpec.hs b/specs/Data/MediaBus/Rtp/SourceSpec.hs --- a/specs/Data/MediaBus/Rtp/SourceSpec.hs +++ b/specs/Data/MediaBus/Rtp/SourceSpec.hs @@ -9,6 +9,7 @@ import Data.Word import Control.Lens import Data.Proxy +import Control.Monad.Logger spec :: Spec spec = rtpSourceSpec >> rtpPayloadDemuxSpec @@ -20,7 +21,8 @@ Start _ -> n + 1 Next _ -> n) 0 - runTestConduit inputs = runConduitPure (sourceList inputs .| + runTestConduit inputs = runNoLogging $ runConduit + (sourceList inputs .| annotateTypeCIn (Proxy :: Proxy (Stream Int Int Int () B.ByteString)) rtpSource .| consume) @@ -124,10 +126,15 @@ length (runTestConduit inputs) `shouldBe` 8 +runNoLogging :: NoLoggingT Identity a -> a +runNoLogging = runIdentity . runNoLoggingT + rtpPayloadDemuxSpec :: Spec rtpPayloadDemuxSpec = describe "rtpPayloadDemux" $ do let runTestConduit inputs payloadHandlers fallback = - runConduitPure (sourceList inputs .| + runNoLogging $ + runConduit + (sourceList inputs .| rtpSource .| rtpPayloadDemux payloadHandlers fallback .| consume) @@ -136,7 +143,7 @@ let inputs = [ MkStream (Start (MkFrameCtx 0 0 0 ())) , mkTestRtpPacket 0 0 0 ] - outs = preview payload <$> runTestConduit inputs [] () + outs = preview eachFrameContent <$> runTestConduit inputs [] () in outs `shouldBe` [ Nothing, Just () ] it "always yields the fallback element if the payload table contains no handler for the payload type" $ @@ -146,11 +153,11 @@ 0 (mkTestPayload (MkRtpPayloadType p)) | p <- [0 .. 128] ] - fallback :: SampleBuffer (S16 8000) + fallback :: Audio (Hz 8000) Mono (Raw S16) fallback = mempty - outs = preview payload <$> runTestConduit inputs + outs = preview eachFrameContent <$> runTestConduit inputs [ ( MkRtpPayloadType 129 - , transcode . + , over (framePayload . eachChannel) decodeALawSample . alawPayloadHandler ) ] @@ -164,21 +171,21 @@ [ mkTestRtpPacketWithPayload 0 0 0 (mkTestPayload 8) , mkTestRtpPacketWithPayload 0 0 0 (mkTestPayload 0) ] - outputs = preview payload <$> runTestConduit inputs + outputs = preview eachFrameContent <$> runTestConduit inputs [ ( 8 - , payload .~ + , framePayload .~ "first 8 handler" ) , ( 8 - , payload .~ + , framePayload .~ "second 8 handler" ) , ( 0 - , payload .~ + , framePayload .~ "first 0 handler" ) , ( 0 - , payload .~ + , framePayload .~ "second 0 handler" ) ] @@ -191,7 +198,7 @@ mkBrokenTestRtpPacket = MkStream (Next (MkFrame 0 0 (B.pack [ 0, 0, 0 ]))) mkTestPayload :: RtpPayloadType -> RtpPayload -mkTestPayload pt = MkRtpPayload pt (sampleBufferFromList [ 0, 0, 0 ]) +mkTestPayload pt = MkRtpPayload pt (mediaBufferFromList [ 0, 0, 0 ]) mkTestRtpPacket :: RtpSsrc -> RtpSeqNum diff --git a/src/Data/MediaBus/Rtp/AlawSource.hs b/src/Data/MediaBus/Rtp/AlawSource.hs --- a/src/Data/MediaBus/Rtp/AlawSource.hs +++ b/src/Data/MediaBus/Rtp/AlawSource.hs @@ -1,39 +1,46 @@ -module Data.MediaBus.Rtp.AlawSource ( rtpAlaw16kHzS16Source ) where +-- | A high-level RTP receiver for G.711-ALAW. +module Data.MediaBus.Rtp.AlawSource + ( rtpAlaw16kHzS16Source + , alawPayloadHandler + ) 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 ) +import Conduit +import qualified Data.ByteString as B +import Data.MediaBus +import Data.MediaBus.Rtp.Packet +import Data.MediaBus.Rtp.Source +import Data.Proxy +import Data.Streaming.Network (HostPreference) +import Data.Word +import Network.Socket (SockAddr) +import Control.Lens +import Data.Coerce +import Control.Monad.Logger -rtpAlaw16kHzS16Source :: MonadResource m - => Int - -> HostPreference - -> Int - -> Source m (Stream RtpSsrc RtpSeqNum (Ticks 16000 Word64) () (Segment (640 :@ 16000) (SampleBuffer (S16 16000)))) +-- | Opend a UDP port and listen for RTP- G711 Alaw packets +rtpAlaw16kHzS16Source + :: (MonadLogger m, MonadResource m) + => Int + -> HostPreference + -> Int + -> Source m (Stream RtpSsrc RtpSeqNum (Ticks (Hz 16000) Word64) () (Segment (640 :/ Hz 16000) (Audio (Hz 16000) Mono (Raw S16)))) 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 + annotateTypeSource + (Proxy :: Proxy (Stream (SourceId (Maybe SockAddr)) RtpSeqNum (ClockTimeDiff UtcClock) () B.ByteString)) + (udpDatagramSource useUtcClock udpListenPort udpListenIP) .| + rtpSource .| + rtpPayloadDemux [(8, alawPayloadHandler)] mempty .| + annotateTypeCOut + (Proxy :: Proxy (Stream RtpSsrc RtpSeqNum (Ticks (Hz 8000) Word32) () (Audio (Hz 8000) Mono (Raw S16)))) + alawToS16 .| + resample8to16kHz' (0 :: Pcm Mono S16) .| + convertTicksC' + (Proxy :: Proxy '( Hz 8000, Word32)) + (Proxy :: Proxy '( Hz 16000, Word64)) .| + reorderFramesBySeqNumC reorderBufferSize .| + segmentC + +-- | Coerce an 'RtpPayload' to an 'ALaw' buffer. +alawPayloadHandler :: RtpPayloadHandler t (Audio (Hz 8000) Mono (Raw ALaw)) +alawPayloadHandler = + framePayload %~ (view (from pcmMediaBuffer) . coerce . _rtpPayload) diff --git a/src/Data/MediaBus/Rtp/Packet.lhs b/src/Data/MediaBus/Rtp/Packet.lhs --- a/src/Data/MediaBus/Rtp/Packet.lhs +++ b/src/Data/MediaBus/Rtp/Packet.lhs @@ -18,10 +18,10 @@ > import Text.Printf > import Data.Word > import Data.Bits -> import Data.MediaBus.Monotone -> import Data.MediaBus.Sequence +> import Data.MediaBus.Basics.Monotone +> import Data.MediaBus.Basics.Sequence > import Control.Lens -> import Data.MediaBus.Sample +> import Data.MediaBus.Media.Buffer > import GHC.Generics ( Generic ) > import Control.DeepSeq > import System.Random @@ -72,8 +72,10 @@ 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 +> (MkRtpTimestamp l) `compare` (MkRtpTimestamp r) +> | l == r = EQ +> | l `succeeds` r = GT +> | otherwise = LT SeqNum numbers are special because they wrap-around. @@ -110,7 +112,7 @@ with the 'RtpPayloadType'. > data RtpPayload = MkRtpPayload { _rtpPayloadType :: RtpPayloadType -> , _rtpPayload :: SampleBuffer Word8 +> , _rtpPayload :: MediaBuffer Word8 > } > deriving (Eq, Generic) @@ -145,7 +147,7 @@ > let bodyBytes = if hasPadding h > then adjustPadding remainingBytes > else remainingBytes -> body = MkRtpPayload pt (sampleBufferFromByteString bodyBytes) +> body = MkRtpPayload pt (mediaBufferFromByteString bodyBytes) Wrap everything up and return it: @@ -364,7 +366,7 @@ packet are listed, allowing correct talker indication at the receiver. -> csrcs' <- sequence (replicate (fromIntegral csrcCount) getWord32be) +> csrcs' <- replicateM (fromIntegral csrcCount) getWord32be If a the extension flag is set, we must parse an optional header extension: @@ -407,7 +409,7 @@ > getHeaderExtension = do > field <- getWord16be > len <- getWord16be -> body <- sequence (replicate (fromIntegral len) getWord32be) +> body <- replicateM (fromIntegral len) getWord32be > return (MkHeaderExtension field body) NOTE: To test this, you can use gstreamer, e.g. with this command line: @@ -446,12 +448,12 @@ First write the header then the body. > putPayloadTypeAndHeader (_rtpPayloadType b) h -> putByteString (byteStringFromSampleBuffer (_rtpPayload b)) +> putByteString (mediaBufferToByteString (_rtpPayload b)) Calculate number of bytes required for padding. > let paddingLen = fromIntegral -> ((64 - ((sampleCount (_rtpPayload b)) `rem` 64)) `rem` 64) +> ((64 - (mediaBufferLength (_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/. @@ -523,7 +525,7 @@ The maximum of 16 csrcs: -> mapM_ putWord32be (rtpSsrc <$> (take 16 csrcs)) +> mapM_ putWord32be (rtpSsrc <$> take 16 csrcs) And last but not least the header extensions: diff --git a/src/Data/MediaBus/Rtp/Source.hs b/src/Data/MediaBus/Rtp/Source.hs --- a/src/Data/MediaBus/Rtp/Source.hs +++ b/src/Data/MediaBus/Rtp/Source.hs @@ -1,117 +1,141 @@ +-- | A conduit 'Source' that receives bytestrings and parses RTP packets. {-# OPTIONS -Wno-unused-top-binds #-} module Data.MediaBus.Rtp.Source - ( type RtpStream - , rtpSource - , rtpPayloadDemux - , type RtpPayloadHandler - , alawPayloadHandler - ) where + ( type RtpStream + , rtpSource + , rtpPayloadDemux + , type RtpPayloadHandler + ) 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 Conduit +import Control.Lens +import Control.Monad +import Control.Monad.Logger +import qualified Data.ByteString as B +import Data.Default import qualified Data.List -import Data.Maybe +import Data.Maybe +import Data.MediaBus +import Data.MediaBus.Rtp.Packet +import Data.String +import Text.Printf +-- | A handy alias for a 'Stream' for RTP packets, with according sequence +-- number, time stamp and payload types. type RtpStream p = Stream RtpSsrc RtpSeqNum RtpTimestamp p RtpPayload -data RRState ctx = MkRRState { _currCtx :: ctx - , _isFirstPacket :: Bool - } - deriving (Show) +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 +-- | A 'Conduit' that consumes 'ByteString's and produces an 'RtpStream' +rtpSource + :: ( Show i + , Show s + , Show t + , Show p + , Default i + , Monad m + , Default p + , Show p + , MonadLogger m + ) + => 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 frm@(MkStream (Start _)) = + $logInfo (fromString ("state frame received: " ++ show frm)) 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)) - + case deserialize contentIn of + Left rtpError -> logRtpError rtpError + Right rtpPacket -> do + let rtpHeader = header rtpPacket + res <- updateState rtpHeader + unless (res == FrameCtxNotChanged) (yieldStreamStart res) + yieldStreamNext (body rtpPacket) + logRtpError e = do + ctx <- use currCtx + $logError (fromString (printf "rtp packet parse error: %s, frame-context: %s" 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 + oldCtx <- + currCtx <<%= + ((frameCtxSeqNumRef .~ sequenceNumber rtpHeader) . + (frameCtxTimestampRef .~ headerTimestamp rtpHeader)) + wasFirstPacket <- isFirstPacket <<.= False + if oldCtx ^. frameCtxSourceId /= ssrc rtpHeader + then do + currCtx . frameCtxSourceId .= ssrc rtpHeader + return RtpSsrcChanged + else if sequenceNumbersDifferTooMuch + (oldCtx ^. frameCtxSeqNumRef) + (sequenceNumber rtpHeader) + then return RtpSequenceNumberGap + else if timestampsDifferTooMuch + (oldCtx ^. frameCtxTimestampRef) + (headerTimestamp rtpHeader) + then return RtpTimestampGap + else if wasFirstPacket + then return NewRtpSession + 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 + 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 + let d = + if currTS >= oldTS + then currTS - oldTS + else oldTS - currTS + timestampMaxDelta = 2000 -- TODO extract + in d >= timestampMaxDelta + yieldStreamStart why = do + fx <- use currCtx + $logDebug (fromString (show why ++ ": starting rtp stream: " ++ show fx)) + yieldStartFrameCtx fx yieldStreamNext p = do - ts <- use (currCtx . frameCtxTimestampRef) - sn <- use (currCtx . frameCtxSeqNumRef) - yield (MkStream (Next (MkFrame ts sn p))) + ts <- use (currCtx . frameCtxTimestampRef) + sn <- use (currCtx . frameCtxSeqNumRef) + yield (MkStream (Next (MkFrame ts sn p))) -data RRSourceChange = FrameCtxChanged | FrameCtxNotChanged - deriving (Eq) +-- | The stream might re-'Start' for these reasons: +data RRSourceChange + = NewRtpSession + | RtpSsrcChanged + | RtpSequenceNumberGap + | RtpTimestampGap + | FrameCtxNotChanged + deriving (Eq, Show) -rtpPayloadDemux :: (Integral t, Monad m) - => [(RtpPayloadType, RtpPayloadHandler (Ticks r t) c)] - -> c - -> Conduit (RtpStream p) m (Stream RtpSsrc RtpSeqNum (Ticks r t) p c) +-- | A utility that call the right 'RtpPayloadHandler' for the 'RtpPayloadType' +-- of the 'Frame'. +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 + mapC (timestamp %~ (MkTicks . fromIntegral . _rtpTimestamp)) .| + awaitForever go where - setFallbackContent = payload .~ fallbackContent + setFallbackContent = framePayload .~ 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 + 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 -alawPayloadHandler :: RtpPayloadHandler t (SampleBuffer ALaw) -alawPayloadHandler = payload %~ (coerce . _rtpPayload) +-- | Functions from 'Frame' to 'Frame' for converting/coercing an 'RtpPayload' +-- to some common media type. NOTE: This is not for transcoding, this is rather +-- for /casting/ or /coercing/ +type RtpPayloadHandler t c = Frame RtpSeqNum t RtpPayload -> Frame RtpSeqNum t c