mediabus-rtp 0.2.0.0 → 0.3.2.0

raw patch · 6 files changed

+208/−165 lines, 6 filesdep +monad-loggerdep ~mediabusdep ~mediabus-rtpPVP ok

version bump matches the API change (PVP)

Dependencies added: monad-logger

Dependency ranges changed: mediabus, mediabus-rtp

API changes (from Hackage documentation)

Files

• mediabus-demo-rtp-alaw-player/Main.hs +6/−6
• mediabus-rtp.cabal +9/−6
• specs/Data/MediaBus/Rtp/SourceSpec.hs +19/−12
• src/Data/MediaBus/Rtp/AlawSource.hs +43/−36
• src/Data/MediaBus/Rtp/Packet.lhs +14/−12
• src/Data/MediaBus/Rtp/Source.hs +117/−93

@@ -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)

@@ -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,

@@ -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

@@ -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)

@@ -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: 

@@ -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