diff options
author | kali kaneko (leap communications) <kali@leap.se> | 2021-11-29 01:46:27 +0100 |
---|---|---|
committer | kali kaneko (leap communications) <kali@leap.se> | 2021-11-29 18:14:16 +0100 |
commit | 18f52af5be3a9a0c73811706108f790d65ee9c67 (patch) | |
tree | e13cbacb47d56919caa9c44a2b45dec1497a7860 /vendor/github.com/pion/interceptor | |
parent | ebcef0d57b6ecb5a40c6579f6be07182dd3033ba (diff) |
[pkg] update vendor
Diffstat (limited to 'vendor/github.com/pion/interceptor')
29 files changed, 1756 insertions, 0 deletions
diff --git a/vendor/github.com/pion/interceptor/.gitignore b/vendor/github.com/pion/interceptor/.gitignore new file mode 100644 index 0000000..83db74b --- /dev/null +++ b/vendor/github.com/pion/interceptor/.gitignore @@ -0,0 +1,24 @@ +### JetBrains IDE ### +##################### +.idea/ + +### Emacs Temporary Files ### +############################# +*~ + +### Folders ### +############### +bin/ +vendor/ +node_modules/ + +### Files ### +############# +*.ivf +*.ogg +tags +cover.out +*.sw[poe] +*.wasm +examples/sfu-ws/cert.pem +examples/sfu-ws/key.pem diff --git a/vendor/github.com/pion/interceptor/.golangci.yml b/vendor/github.com/pion/interceptor/.golangci.yml new file mode 100644 index 0000000..d6162c9 --- /dev/null +++ b/vendor/github.com/pion/interceptor/.golangci.yml @@ -0,0 +1,89 @@ +linters-settings: + govet: + check-shadowing: true + misspell: + locale: US + exhaustive: + default-signifies-exhaustive: true + gomodguard: + blocked: + modules: + - github.com/pkg/errors: + recommendations: + - errors + +linters: + enable: + - asciicheck # Simple linter to check that your code does not contain non-ASCII identifiers + - bodyclose # checks whether HTTP response body is closed successfully + - deadcode # Finds unused code + - depguard # Go linter that checks if package imports are in a list of acceptable packages + - dogsled # Checks assignments with too many blank identifiers (e.g. x, _, _, _, := f()) + - dupl # Tool for code clone detection + - errcheck # Errcheck is a program for checking for unchecked errors in go programs. These unchecked errors can be critical bugs in some cases + - exhaustive # check exhaustiveness of enum switch statements + - exportloopref # checks for pointers to enclosing loop variables + - gci # Gci control golang package import order and make it always deterministic. + - gochecknoglobals # Checks that no globals are present in Go code + - gochecknoinits # Checks that no init functions are present in Go code + - gocognit # Computes and checks the cognitive complexity of functions + - goconst # Finds repeated strings that could be replaced by a constant + - gocritic # The most opinionated Go source code linter + - godox # Tool for detection of FIXME, TODO and other comment keywords + - goerr113 # Golang linter to check the errors handling expressions + - gofmt # Gofmt checks whether code was gofmt-ed. By default this tool runs with -s option to check for code simplification + - gofumpt # Gofumpt checks whether code was gofumpt-ed. + - goheader # Checks is file header matches to pattern + - goimports # Goimports does everything that gofmt does. Additionally it checks unused imports + - golint # Golint differs from gofmt. Gofmt reformats Go source code, whereas golint prints out style mistakes + - gomodguard # Allow and block list linter for direct Go module dependencies. This is different from depguard where there are different block types for example version constraints and module recommendations. + - goprintffuncname # Checks that printf-like functions are named with `f` at the end + - gosec # Inspects source code for security problems + - gosimple # Linter for Go source code that specializes in simplifying a code + - govet # Vet examines Go source code and reports suspicious constructs, such as Printf calls whose arguments do not align with the format string + - ineffassign # Detects when assignments to existing variables are not used + - misspell # Finds commonly misspelled English words in comments + - nakedret # Finds naked returns in functions greater than a specified function length + - noctx # noctx finds sending http request without context.Context + - scopelint # Scopelint checks for unpinned variables in go programs + - staticcheck # Staticcheck is a go vet on steroids, applying a ton of static analysis checks + - structcheck # Finds unused struct fields + - stylecheck # Stylecheck is a replacement for golint + - typecheck # Like the front-end of a Go compiler, parses and type-checks Go code + - unconvert # Remove unnecessary type conversions + - unparam # Reports unused function parameters + - unused # Checks Go code for unused constants, variables, functions and types + - varcheck # Finds unused global variables and constants + - whitespace # Tool for detection of leading and trailing whitespace + disable: + - funlen # Tool for detection of long functions + - gocyclo # Computes and checks the cyclomatic complexity of functions + - godot # Check if comments end in a period + - gomnd # An analyzer to detect magic numbers. + - lll # Reports long lines + - maligned # Tool to detect Go structs that would take less memory if their fields were sorted + - nestif # Reports deeply nested if statements + - nlreturn # nlreturn checks for a new line before return and branch statements to increase code clarity + - nolintlint # Reports ill-formed or insufficient nolint directives + - prealloc # Finds slice declarations that could potentially be preallocated + - rowserrcheck # checks whether Err of rows is checked successfully + - sqlclosecheck # Checks that sql.Rows and sql.Stmt are closed. + - testpackage # linter that makes you use a separate _test package + - wsl # Whitespace Linter - Forces you to use empty lines! + +issues: + exclude-use-default: false + exclude-rules: + # Allow complex tests, better to be self contained + - path: _test\.go + linters: + - gocognit + + # Allow complex main function in examples + - path: examples + text: "of func `main` is high" + linters: + - gocognit + +run: + skip-dirs-use-default: false diff --git a/vendor/github.com/pion/interceptor/LICENSE b/vendor/github.com/pion/interceptor/LICENSE new file mode 100644 index 0000000..ab60297 --- /dev/null +++ b/vendor/github.com/pion/interceptor/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2018 + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/vendor/github.com/pion/interceptor/README.md b/vendor/github.com/pion/interceptor/README.md new file mode 100644 index 0000000..007c340 --- /dev/null +++ b/vendor/github.com/pion/interceptor/README.md @@ -0,0 +1,81 @@ +<h1 align="center"> + <br> + Pion Interceptor + <br> +</h1> +<h4 align="center">RTCP and RTCP processors for building real time communications</h4> +<p align="center"> + <a href="https://pion.ly"><img src="https://img.shields.io/badge/pion-interceptor-gray.svg?longCache=true&colorB=brightgreen" alt="Pion Interceptor"></a> + <a href="https://pion.ly/slack"><img src="https://img.shields.io/badge/join-us%20on%20slack-gray.svg?longCache=true&logo=slack&colorB=brightgreen" alt="Slack Widget"></a> + <br> + <a href="https://pkg.go.dev/github.com/pion/interceptor"><img src="https://godoc.org/github.com/pion/interceptor?status.svg" alt="GoDoc"></a> + <a href="https://codecov.io/gh/pion/interceptor"><img src="https://codecov.io/gh/pion/interceptor/branch/master/graph/badge.svg" alt="Coverage Status"></a> + <a href="https://goreportcard.com/report/github.com/pion/interceptor"><img src="https://goreportcard.com/badge/github.com/pion/interceptor" alt="Go Report Card"></a> + <a href="LICENSE"><img src="https://img.shields.io/badge/License-MIT-yellow.svg" alt="License: MIT"></a> +</p> +<br> + +Interceptor is a framework for building RTP/RTCP communication software. This framework defines +a interface that each interceptor must satisfy. These interceptors are then run sequentially. We +also then provide common interceptors that will be useful for building RTC software. + +This package was built for [pion/webrtc](https://github.com/pion/webrtc), but we designed it to be consumable +by anyone. With the following tenets in mind. + +* Useful defaults. Each interceptor will be configured to give you a good default experience. +* Unblock unique use cases. New use cases are what is driving WebRTC, we want to empower them. +* Encourage modification. Add your own interceptors without forking. Mixing with the ones we provide. +* Empower learning. This code base should be useful to read and learn even if you aren't using Pion. + +#### Current Interceptors +* NACK Generator/Responder + +#### Planned Interceptors +* [Sender and Receiver Reports](https://tools.ietf.org/html/rfc3550#section-6.4) + - Bandwidth Estimation from Receiver Reports +* [Transport Wide Congestion Control Feedback](https://tools.ietf.org/html/draft-holmer-rmcat-transport-wide-cc-extensions-01) + - [NADA](https://tools.ietf.org/html/rfc8698) + - [Google Congestion Control](https://tools.ietf.org/html/draft-ietf-rmcat-gcc-02) +* JitterBuffer, re-order packets and wait for arrival +* [FlexFec](https://tools.ietf.org/html/draft-ietf-payload-flexible-fec-scheme-20) +* [webrtc-stats](https://www.w3.org/TR/webrtc-stats/) compliant statistics generation + +### Interceptor Public API +The public interface is defined in [interceptor.go](https://github.com/pion/interceptor/blob/master/interceptor.go). +The methods you need to satisy are broken up into 4 groups. + +* `BindRTCPWriter` and `BindRTCPReader` allow you to inspect/modify RTCP traffic. +* `BindLocalStream` and `BindRemoteStream` notify you of a new SSRC stream and allow you to inspect/modify. +* `UnbindLocalStream` and `UnbindRemoteStream` notify you when a SSRC stream has been removed +* `Close` called when the interceptor is closed. + +Interceptors also pass Attributes between each other. These are a collection of key/value pairs and are useful for storing metadata +or caching. + +[noop.go](https://github.com/pion/interceptor/blob/master/noop.go) is an interceptor that satisfies this interface, but does nothing. +You can embed this interceptor as a starting point so you only need to define exactly what you need. + +[chain.go]( https://github.com/pion/interceptor/blob/master/chain.go) is used to combine multiple interceptors into one. They are called +sequentially as the packet moves through them. + +### Examples +The [examples](https://github.com/pion/interceptor/blob/master/examples) directory provides some basic examples. If you need more please file an issue! +You should also look in [pion/webrtc](https://github.com/pion/webrtc) for real world examples. + +### Community +Pion has an active community on the [Golang Slack](https://invite.slack.golangbridge.org/). Sign up and join the **#pion** channel for discussions and support. You can also use [Pion mailing list](https://groups.google.com/forum/#!forum/pion). + +We are always looking to support **your projects**. Please reach out if you have something to build! + +If you need commercial support or don't want to use public methods you can contact us at [team@pion.ly](mailto:team@pion.ly) + +### Contributing +Check out the **[contributing wiki](https://github.com/pion/webrtc/wiki/Contributing)** to join the group of amazing people making this project possible: + +* [Adam Kiss](https://github.com/masterada) - *Original Author* +* [Sean DuBois](https://github.com/sean-der) - *Original Author* +* [Atsushi Watanabe](https://github.com/at-wat) +* [Alessandro Ros](https://github.com/aler9) + +### License +MIT License - see [LICENSE](LICENSE) for full text diff --git a/vendor/github.com/pion/interceptor/chain.go b/vendor/github.com/pion/interceptor/chain.go new file mode 100644 index 0000000..d53c307 --- /dev/null +++ b/vendor/github.com/pion/interceptor/chain.go @@ -0,0 +1,75 @@ +package interceptor + +// Chain is an interceptor that runs all child interceptors in order. +type Chain struct { + interceptors []Interceptor +} + +// NewChain returns a new Chain interceptor. +func NewChain(interceptors []Interceptor) *Chain { + return &Chain{interceptors: interceptors} +} + +// BindRTCPReader lets you modify any incoming RTCP packets. It is called once per sender/receiver, however this might +// change in the future. The returned method will be called once per packet batch. +func (i *Chain) BindRTCPReader(reader RTCPReader) RTCPReader { + for _, interceptor := range i.interceptors { + reader = interceptor.BindRTCPReader(reader) + } + + return reader +} + +// BindRTCPWriter lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method +// will be called once per packet batch. +func (i *Chain) BindRTCPWriter(writer RTCPWriter) RTCPWriter { + for _, interceptor := range i.interceptors { + writer = interceptor.BindRTCPWriter(writer) + } + + return writer +} + +// BindLocalStream lets you modify any outgoing RTP packets. It is called once for per LocalStream. The returned method +// will be called once per rtp packet. +func (i *Chain) BindLocalStream(ctx *StreamInfo, writer RTPWriter) RTPWriter { + for _, interceptor := range i.interceptors { + writer = interceptor.BindLocalStream(ctx, writer) + } + + return writer +} + +// UnbindLocalStream is called when the Stream is removed. It can be used to clean up any data related to that track. +func (i *Chain) UnbindLocalStream(ctx *StreamInfo) { + for _, interceptor := range i.interceptors { + interceptor.UnbindLocalStream(ctx) + } +} + +// BindRemoteStream lets you modify any incoming RTP packets. It is called once for per RemoteStream. The returned method +// will be called once per rtp packet. +func (i *Chain) BindRemoteStream(ctx *StreamInfo, reader RTPReader) RTPReader { + for _, interceptor := range i.interceptors { + reader = interceptor.BindRemoteStream(ctx, reader) + } + + return reader +} + +// UnbindRemoteStream is called when the Stream is removed. It can be used to clean up any data related to that track. +func (i *Chain) UnbindRemoteStream(ctx *StreamInfo) { + for _, interceptor := range i.interceptors { + interceptor.UnbindRemoteStream(ctx) + } +} + +// Close closes the Interceptor, cleaning up any data if necessary. +func (i *Chain) Close() error { + var errs []error + for _, interceptor := range i.interceptors { + errs = append(errs, interceptor.Close()) + } + + return flattenErrs(errs) +} diff --git a/vendor/github.com/pion/interceptor/codecov.yml b/vendor/github.com/pion/interceptor/codecov.yml new file mode 100644 index 0000000..085200a --- /dev/null +++ b/vendor/github.com/pion/interceptor/codecov.yml @@ -0,0 +1,20 @@ +# +# DO NOT EDIT THIS FILE +# +# It is automatically copied from https://github.com/pion/.goassets repository. +# + +coverage: + status: + project: + default: + # Allow decreasing 2% of total coverage to avoid noise. + threshold: 2% + patch: + default: + target: 70% + only_pulls: true + +ignore: + - "examples/*" + - "examples/**/*" diff --git a/vendor/github.com/pion/interceptor/errors.go b/vendor/github.com/pion/interceptor/errors.go new file mode 100644 index 0000000..45e1252 --- /dev/null +++ b/vendor/github.com/pion/interceptor/errors.go @@ -0,0 +1,51 @@ +package interceptor + +import ( + "errors" + "strings" +) + +func flattenErrs(errs []error) error { + errs2 := []error{} + for _, e := range errs { + if e != nil { + errs2 = append(errs2, e) + } + } + if len(errs2) == 0 { + return nil + } + return multiError(errs2) +} + +type multiError []error + +func (me multiError) Error() string { + var errstrings []string + + for _, err := range me { + if err != nil { + errstrings = append(errstrings, err.Error()) + } + } + + if len(errstrings) == 0 { + return "multiError must contain multiple error but is empty" + } + + return strings.Join(errstrings, "\n") +} + +func (me multiError) Is(err error) bool { + for _, e := range me { + if errors.Is(e, err) { + return true + } + if me2, ok := e.(multiError); ok { + if me2.Is(err) { + return true + } + } + } + return false +} diff --git a/vendor/github.com/pion/interceptor/go.mod b/vendor/github.com/pion/interceptor/go.mod new file mode 100644 index 0000000..1370bc9 --- /dev/null +++ b/vendor/github.com/pion/interceptor/go.mod @@ -0,0 +1,10 @@ +module github.com/pion/interceptor + +go 1.15 + +require ( + github.com/pion/logging v0.2.2 + github.com/pion/rtcp v1.2.6 + github.com/pion/rtp v1.6.2 + github.com/stretchr/testify v1.7.0 +) diff --git a/vendor/github.com/pion/interceptor/go.sum b/vendor/github.com/pion/interceptor/go.sum new file mode 100644 index 0000000..104d018 --- /dev/null +++ b/vendor/github.com/pion/interceptor/go.sum @@ -0,0 +1,21 @@ +github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY= +github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms= +github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA= +github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8= +github.com/pion/rtcp v1.2.6 h1:1zvwBbyd0TeEuuWftrd/4d++m+/kZSeiguxU61LFWpo= +github.com/pion/rtcp v1.2.6/go.mod h1:52rMNPWFsjr39z9B9MhnkqhPLoeHTv1aN63o/42bWE0= +github.com/pion/rtp v1.6.2 h1:iGBerLX6JiDjB9NXuaPzHyxHFG9JsIEdgwTC0lp5n/U= +github.com/pion/rtp v1.6.2/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0= +github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/vendor/github.com/pion/interceptor/interceptor.go b/vendor/github.com/pion/interceptor/interceptor.go new file mode 100644 index 0000000..2d1b97e --- /dev/null +++ b/vendor/github.com/pion/interceptor/interceptor.go @@ -0,0 +1,108 @@ +// Package interceptor contains the Interceptor interface, with some useful interceptors that should be safe to use +// in most cases. +package interceptor + +import ( + "io" + + "github.com/pion/rtcp" + "github.com/pion/rtp" +) + +// Interceptor can be used to add functionality to you PeerConnections by modifying any incoming/outgoing rtp/rtcp +// packets, or sending your own packets as needed. +type Interceptor interface { + + // BindRTCPReader lets you modify any incoming RTCP packets. It is called once per sender/receiver, however this might + // change in the future. The returned method will be called once per packet batch. + BindRTCPReader(reader RTCPReader) RTCPReader + + // BindRTCPWriter lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method + // will be called once per packet batch. + BindRTCPWriter(writer RTCPWriter) RTCPWriter + + // BindLocalStream lets you modify any outgoing RTP packets. It is called once for per LocalStream. The returned method + // will be called once per rtp packet. + BindLocalStream(info *StreamInfo, writer RTPWriter) RTPWriter + + // UnbindLocalStream is called when the Stream is removed. It can be used to clean up any data related to that track. + UnbindLocalStream(info *StreamInfo) + + // BindRemoteStream lets you modify any incoming RTP packets. It is called once for per RemoteStream. The returned method + // will be called once per rtp packet. + BindRemoteStream(info *StreamInfo, reader RTPReader) RTPReader + + // UnbindRemoteStream is called when the Stream is removed. It can be used to clean up any data related to that track. + UnbindRemoteStream(info *StreamInfo) + + io.Closer +} + +// RTPWriter is used by Interceptor.BindLocalStream. +type RTPWriter interface { + // Write a rtp packet + Write(header *rtp.Header, payload []byte, attributes Attributes) (int, error) +} + +// RTPReader is used by Interceptor.BindRemoteStream. +type RTPReader interface { + // Read a rtp packet + Read([]byte, Attributes) (int, Attributes, error) +} + +// RTCPWriter is used by Interceptor.BindRTCPWriter. +type RTCPWriter interface { + // Write a batch of rtcp packets + Write(pkts []rtcp.Packet, attributes Attributes) (int, error) +} + +// RTCPReader is used by Interceptor.BindRTCPReader. +type RTCPReader interface { + // Read a batch of rtcp packets + Read([]byte, Attributes) (int, Attributes, error) +} + +// Attributes are a generic key/value store used by interceptors +type Attributes map[interface{}]interface{} + +// RTPWriterFunc is an adapter for RTPWrite interface +type RTPWriterFunc func(header *rtp.Header, payload []byte, attributes Attributes) (int, error) + +// RTPReaderFunc is an adapter for RTPReader interface +type RTPReaderFunc func([]byte, Attributes) (int, Attributes, error) + +// RTCPWriterFunc is an adapter for RTCPWriter interface +type RTCPWriterFunc func(pkts []rtcp.Packet, attributes Attributes) (int, error) + +// RTCPReaderFunc is an adapter for RTCPReader interface +type RTCPReaderFunc func([]byte, Attributes) (int, Attributes, error) + +// Write a rtp packet +func (f RTPWriterFunc) Write(header *rtp.Header, payload []byte, attributes Attributes) (int, error) { + return f(header, payload, attributes) +} + +// Read a rtp packet +func (f RTPReaderFunc) Read(b []byte, a Attributes) (int, Attributes, error) { + return f(b, a) +} + +// Write a batch of rtcp packets +func (f RTCPWriterFunc) Write(pkts []rtcp.Packet, attributes Attributes) (int, error) { + return f(pkts, attributes) +} + +// Read a batch of rtcp packets +func (f RTCPReaderFunc) Read(b []byte, a Attributes) (int, Attributes, error) { + return f(b, a) +} + +// Get returns the attribute associated with key. +func (a Attributes) Get(key interface{}) interface{} { + return a[key] +} + +// Set sets the attribute associated with key to the given value. +func (a Attributes) Set(key interface{}, val interface{}) { + a[key] = val +} diff --git a/vendor/github.com/pion/interceptor/noop.go b/vendor/github.com/pion/interceptor/noop.go new file mode 100644 index 0000000..2dc4e8e --- /dev/null +++ b/vendor/github.com/pion/interceptor/noop.go @@ -0,0 +1,40 @@ +package interceptor + +// NoOp is an Interceptor that does not modify any packets. It can embedded in other interceptors, so it's +// possible to implement only a subset of the methods. +type NoOp struct{} + +// BindRTCPReader lets you modify any incoming RTCP packets. It is called once per sender/receiver, however this might +// change in the future. The returned method will be called once per packet batch. +func (i *NoOp) BindRTCPReader(reader RTCPReader) RTCPReader { + return reader +} + +// BindRTCPWriter lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method +// will be called once per packet batch. +func (i *NoOp) BindRTCPWriter(writer RTCPWriter) RTCPWriter { + return writer +} + +// BindLocalStream lets you modify any outgoing RTP packets. It is called once for per LocalStream. The returned method +// will be called once per rtp packet. +func (i *NoOp) BindLocalStream(_ *StreamInfo, writer RTPWriter) RTPWriter { + return writer +} + +// UnbindLocalStream is called when the Stream is removed. It can be used to clean up any data related to that track. +func (i *NoOp) UnbindLocalStream(_ *StreamInfo) {} + +// BindRemoteStream lets you modify any incoming RTP packets. It is called once for per RemoteStream. The returned method +// will be called once per rtp packet. +func (i *NoOp) BindRemoteStream(_ *StreamInfo, reader RTPReader) RTPReader { + return reader +} + +// UnbindRemoteStream is called when the Stream is removed. It can be used to clean up any data related to that track. +func (i *NoOp) UnbindRemoteStream(_ *StreamInfo) {} + +// Close closes the Interceptor, cleaning up any data if necessary. +func (i *NoOp) Close() error { + return nil +} diff --git a/vendor/github.com/pion/interceptor/pkg/nack/errors.go b/vendor/github.com/pion/interceptor/pkg/nack/errors.go new file mode 100644 index 0000000..bbfc773 --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/nack/errors.go @@ -0,0 +1,6 @@ +package nack + +import "errors" + +// ErrInvalidSize is returned by newReceiveLog/newSendBuffer, when an incorrect buffer size is supplied. +var ErrInvalidSize = errors.New("invalid buffer size") diff --git a/vendor/github.com/pion/interceptor/pkg/nack/generator_interceptor.go b/vendor/github.com/pion/interceptor/pkg/nack/generator_interceptor.go new file mode 100644 index 0000000..447a949 --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/nack/generator_interceptor.go @@ -0,0 +1,162 @@ +package nack + +import ( + "math/rand" + "sync" + "time" + + "github.com/pion/interceptor" + "github.com/pion/logging" + "github.com/pion/rtcp" + "github.com/pion/rtp" +) + +// GeneratorInterceptor interceptor generates nack feedback messages. +type GeneratorInterceptor struct { + interceptor.NoOp + size uint16 + skipLastN uint16 + interval time.Duration + m sync.Mutex + wg sync.WaitGroup + close chan struct{} + log logging.LeveledLogger + + receiveLogs map[uint32]*receiveLog + receiveLogsMu sync.Mutex +} + +// NewGeneratorInterceptor returns a new GeneratorInterceptor interceptor +func NewGeneratorInterceptor(opts ...GeneratorOption) (*GeneratorInterceptor, error) { + r := &GeneratorInterceptor{ + size: 8192, + skipLastN: 0, + interval: time.Millisecond * 100, + receiveLogs: map[uint32]*receiveLog{}, + close: make(chan struct{}), + log: logging.NewDefaultLoggerFactory().NewLogger("nack_generator"), + } + + for _, opt := range opts { + if err := opt(r); err != nil { + return nil, err + } + } + + if _, err := newReceiveLog(r.size); err != nil { + return nil, err + } + + return r, nil +} + +// BindRTCPWriter lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method +// will be called once per packet batch. +func (n *GeneratorInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter { + n.m.Lock() + defer n.m.Unlock() + + if n.isClosed() { + return writer + } + + n.wg.Add(1) + + go n.loop(writer) + + return writer +} + +// BindRemoteStream lets you modify any incoming RTP packets. It is called once for per RemoteStream. The returned method +// will be called once per rtp packet. +func (n *GeneratorInterceptor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader { + if !streamSupportNack(info) { + return reader + } + + // error is already checked in NewGeneratorInterceptor + receiveLog, _ := newReceiveLog(n.size) + n.receiveLogsMu.Lock() + n.receiveLogs[info.SSRC] = receiveLog + n.receiveLogsMu.Unlock() + + return interceptor.RTPReaderFunc(func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) { + i, attr, err := reader.Read(b, a) + if err != nil { + return 0, nil, err + } + + pkt := rtp.Packet{} + if err = pkt.Unmarshal(b[:i]); err != nil { + return 0, nil, err + } + receiveLog.add(pkt.Header.SequenceNumber) + + return i, attr, nil + }) +} + +// UnbindLocalStream is called when the Stream is removed. It can be used to clean up any data related to that track. +func (n *GeneratorInterceptor) UnbindLocalStream(info *interceptor.StreamInfo) { + n.receiveLogsMu.Lock() + delete(n.receiveLogs, info.SSRC) + n.receiveLogsMu.Unlock() +} + +// Close closes the interceptor +func (n *GeneratorInterceptor) Close() error { + defer n.wg.Wait() + n.m.Lock() + defer n.m.Unlock() + + if !n.isClosed() { + close(n.close) + } + + return nil +} + +func (n *GeneratorInterceptor) loop(rtcpWriter interceptor.RTCPWriter) { + defer n.wg.Done() + + senderSSRC := rand.Uint32() // #nosec + + ticker := time.NewTicker(n.interval) + for { + select { + case <-ticker.C: + func() { + n.receiveLogsMu.Lock() + defer n.receiveLogsMu.Unlock() + + for ssrc, receiveLog := range n.receiveLogs { + missing := receiveLog.missingSeqNumbers(n.skipLastN) + if len(missing) == 0 { + continue + } + + nack := &rtcp.TransportLayerNack{ + SenderSSRC: senderSSRC, + MediaSSRC: ssrc, + Nacks: rtcp.NackPairsFromSequenceNumbers(missing), + } + + if _, err := rtcpWriter.Write([]rtcp.Packet{nack}, interceptor.Attributes{}); err != nil { + n.log.Warnf("failed sending nack: %+v", err) + } + } + }() + case <-n.close: + return + } + } +} + +func (n *GeneratorInterceptor) isClosed() bool { + select { + case <-n.close: + return true + default: + return false + } +} diff --git a/vendor/github.com/pion/interceptor/pkg/nack/generator_option.go b/vendor/github.com/pion/interceptor/pkg/nack/generator_option.go new file mode 100644 index 0000000..092f5db --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/nack/generator_option.go @@ -0,0 +1,44 @@ +package nack + +import ( + "time" + + "github.com/pion/logging" +) + +// GeneratorOption can be used to configure GeneratorInterceptor +type GeneratorOption func(r *GeneratorInterceptor) error + +// GeneratorSize sets the size of the interceptor. +// Size must be one of: 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768 +func GeneratorSize(size uint16) GeneratorOption { + return func(r *GeneratorInterceptor) error { + r.size = size + return nil + } +} + +// GeneratorSkipLastN sets the number of packets (n-1 packets before the last received packets) to ignore when generating +// nack requests. +func GeneratorSkipLastN(skipLastN uint16) GeneratorOption { + return func(r *GeneratorInterceptor) error { + r.skipLastN = skipLastN + return nil + } +} + +// GeneratorLog sets a logger for the interceptor +func GeneratorLog(log logging.LeveledLogger) GeneratorOption { + return func(r *GeneratorInterceptor) error { + r.log = log + return nil + } +} + +// GeneratorInterval sets the nack send interval for the interceptor +func GeneratorInterval(interval time.Duration) GeneratorOption { + return func(r *GeneratorInterceptor) error { + r.interval = interval + return nil + } +} diff --git a/vendor/github.com/pion/interceptor/pkg/nack/nack.go b/vendor/github.com/pion/interceptor/pkg/nack/nack.go new file mode 100644 index 0000000..a658e7f --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/nack/nack.go @@ -0,0 +1,14 @@ +// Package nack provides interceptors to implement sending and receiving negative acknowledgements +package nack + +import "github.com/pion/interceptor" + +func streamSupportNack(info *interceptor.StreamInfo) bool { + for _, fb := range info.RTCPFeedback { + if fb.Type == "nack" && fb.Parameter == "" { + return true + } + } + + return false +} diff --git a/vendor/github.com/pion/interceptor/pkg/nack/receive_log.go b/vendor/github.com/pion/interceptor/pkg/nack/receive_log.go new file mode 100644 index 0000000..8107f59 --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/nack/receive_log.go @@ -0,0 +1,134 @@ +package nack + +import ( + "fmt" + "sync" +) + +type receiveLog struct { + packets []uint64 + size uint16 + end uint16 + started bool + lastConsecutive uint16 + m sync.RWMutex +} + +func newReceiveLog(size uint16) (*receiveLog, error) { + allowedSizes := make([]uint16, 0) + correctSize := false + for i := 6; i < 16; i++ { + if size == 1<<i { + correctSize = true + break + } + allowedSizes = append(allowedSizes, 1<<i) + } + + if !correctSize { + return nil, fmt.Errorf("%w: %d is not a valid size, allowed sizes: %v", ErrInvalidSize, size, allowedSizes) + } + + return &receiveLog{ + packets: make([]uint64, size/64), + size: size, + }, nil +} + +func (s *receiveLog) add(seq uint16) { + s.m.Lock() + defer s.m.Unlock() + + if !s.started { + s.setReceived(seq) + s.end = seq + s.started = true + s.lastConsecutive = seq + return + } + + diff := seq - s.end + switch { + case diff == 0: + return + case diff < uint16SizeHalf: + // this means a positive diff, in other words seq > end (with counting for rollovers) + for i := s.end + 1; i != seq; i++ { + // clear packets between end and seq (these may contain packets from a "size" ago) + s.delReceived(i) + } + s.end = seq + + if s.lastConsecutive+1 == seq { + s.lastConsecutive = seq + } else if seq-s.lastConsecutive > s.size { + s.lastConsecutive = seq - s.size + s.fixLastConsecutive() // there might be valid packets at the beginning of the buffer now + } + case s.lastConsecutive+1 == seq: + // negative diff, seq < end (with counting for rollovers) + s.lastConsecutive = seq + s.fixLastConsecutive() // there might be other valid packets after seq + } + + s.setReceived(seq) +} + +func (s *receiveLog) get(seq uint16) bool { + s.m.RLock() + defer s.m.RUnlock() + + diff := s.end - seq + if diff >= uint16SizeHalf { + return false + } + + if diff >= s.size { + return false + } + + return s.getReceived(seq) +} + +func (s *receiveLog) missingSeqNumbers(skipLastN uint16) []uint16 { + s.m.RLock() + defer s.m.RUnlock() + + until := s.end - skipLastN + if until-s.lastConsecutive >= uint16SizeHalf { + // until < s.lastConsecutive (counting for rollover) + return nil + } + + missingPacketSeqNums := make([]uint16, 0) + for i := s.lastConsecutive + 1; i != until+1; i++ { + if !s.getReceived(i) { + missingPacketSeqNums = append(missingPacketSeqNums, i) + } + } + + return missingPacketSeqNums +} + +func (s *receiveLog) setReceived(seq uint16) { + pos := seq % s.size + s.packets[pos/64] |= 1 << (pos % 64) +} + +func (s *receiveLog) delReceived(seq uint16) { + pos := seq % s.size + s.packets[pos/64] &^= 1 << (pos % 64) +} + +func (s *receiveLog) getReceived(seq uint16) bool { + pos := seq % s.size + return (s.packets[pos/64] & (1 << (pos % 64))) != 0 +} + +func (s *receiveLog) fixLastConsecutive() { + i := s.lastConsecutive + 1 + for ; i != s.end+1 && s.getReceived(i); i++ { + // find all consecutive packets + } + s.lastConsecutive = i - 1 +} diff --git a/vendor/github.com/pion/interceptor/pkg/nack/responder_interceptor.go b/vendor/github.com/pion/interceptor/pkg/nack/responder_interceptor.go new file mode 100644 index 0000000..121657e --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/nack/responder_interceptor.go @@ -0,0 +1,119 @@ +package nack + +import ( + "sync" + + "github.com/pion/interceptor" + "github.com/pion/logging" + "github.com/pion/rtcp" + "github.com/pion/rtp" +) + +// ResponderInterceptor responds to nack feedback messages +type ResponderInterceptor struct { + interceptor.NoOp + size uint16 + log logging.LeveledLogger + + streams map[uint32]*localStream + streamsMu sync.Mutex +} + +type localStream struct { + sendBuffer *sendBuffer + rtpWriter interceptor.RTPWriter +} + +// NewResponderInterceptor returns a new GeneratorInterceptor interceptor +func NewResponderInterceptor(opts ...ResponderOption) (*ResponderInterceptor, error) { + r := &ResponderInterceptor{ + size: 8192, + log: logging.NewDefaultLoggerFactory().NewLogger("nack_responder"), + streams: map[uint32]*localStream{}, + } + + for _, opt := range opts { + if err := opt(r); err != nil { + return nil, err + } + } + + if _, err := newSendBuffer(r.size); err != nil { + return nil, err + } + + return r, nil +} + +// BindRTCPReader lets you modify any incoming RTCP packets. It is called once per sender/receiver, however this might +// change in the future. The returned method will be called once per packet batch. +func (n *ResponderInterceptor) BindRTCPReader(reader interceptor.RTCPReader) interceptor.RTCPReader { + return interceptor.RTCPReaderFunc(func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) { + i, attr, err := reader.Read(b, a) + if err != nil { + return 0, nil, err + } + + pkts, err := rtcp.Unmarshal(b[:i]) + if err != nil { + return 0, nil, err + } + for _, rtcpPacket := range pkts { + nack, ok := rtcpPacket.(*rtcp.TransportLayerNack) + if !ok { + continue + } + + go n.resendPackets(nack) + } + + return i, attr, err + }) +} + +// BindLocalStream lets you modify any outgoing RTP packets. It is called once for per LocalStream. The returned method +// will be called once per rtp packet. +func (n *ResponderInterceptor) BindLocalStream(info *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter { + if !streamSupportNack(info) { + return writer + } + + // error is already checked in NewGeneratorInterceptor + sendBuffer, _ := newSendBuffer(n.size) + n.streamsMu.Lock() + n.streams[info.SSRC] = &localStream{sendBuffer: sendBuffer, rtpWriter: writer} + n.streamsMu.Unlock() + + return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) { + sendBuffer.add(&rtp.Packet{Header: *header, Payload: payload}) + return writer.Write(header, payload, attributes) + }) +} + +// UnbindLocalStream is called when the Stream is removed. It can be used to clean up any data related to that track. +func (n *ResponderInterceptor) UnbindLocalStream(info *interceptor.StreamInfo) { + n.streamsMu.Lock() + delete(n.streams, info.SSRC) + n.streamsMu.Unlock() +} + +func (n *ResponderInterceptor) resendPackets(nack *rtcp.TransportLayerNack) { + n.streamsMu.Lock() + stream, ok := n.streams[nack.MediaSSRC] + n.streamsMu.Unlock() + if !ok { + return + } + + for i := range nack.Nacks { + nack.Nacks[i].Range(func(seq uint16) bool { + if p := stream.sendBuffer.get(seq); p != nil { + if _, err := stream.rtpWriter.Write(&p.Header, p.Payload, interceptor.Attributes{}); err != nil { + n.log.Warnf("failed resending nacked packet: %+v", err) + } + } + + return true + }) + } +} diff --git a/vendor/github.com/pion/interceptor/pkg/nack/responder_option.go b/vendor/github.com/pion/interceptor/pkg/nack/responder_option.go new file mode 100644 index 0000000..7ad52c8 --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/nack/responder_option.go @@ -0,0 +1,23 @@ +package nack + +import "github.com/pion/logging" + +// ResponderOption can be used to configure ResponderInterceptor +type ResponderOption func(s *ResponderInterceptor) error + +// ResponderSize sets the size of the interceptor. +// Size must be one of: 1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768 +func ResponderSize(size uint16) ResponderOption { + return func(r *ResponderInterceptor) error { + r.size = size + return nil + } +} + +// ResponderLog sets a logger for the interceptor +func ResponderLog(log logging.LeveledLogger) ResponderOption { + return func(r *ResponderInterceptor) error { + r.log = log + return nil + } +} diff --git a/vendor/github.com/pion/interceptor/pkg/nack/send_buffer.go b/vendor/github.com/pion/interceptor/pkg/nack/send_buffer.go new file mode 100644 index 0000000..cf3f020 --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/nack/send_buffer.go @@ -0,0 +1,74 @@ +package nack + +import ( + "fmt" + + "github.com/pion/rtp" +) + +const ( + uint16SizeHalf = 1 << 15 +) + +type sendBuffer struct { + packets []*rtp.Packet + size uint16 + lastAdded uint16 + started bool +} + +func newSendBuffer(size uint16) (*sendBuffer, error) { + allowedSizes := make([]uint16, 0) + correctSize := false + for i := 0; i < 16; i++ { + if size == 1<<i { + correctSize = true + break + } + allowedSizes = append(allowedSizes, 1<<i) + } + + if !correctSize { + return nil, fmt.Errorf("%w: %d is not a valid size, allowed sizes: %v", ErrInvalidSize, size, allowedSizes) + } + + return &sendBuffer{ + packets: make([]*rtp.Packet, size), + size: size, + }, nil +} + +func (s *sendBuffer) add(packet *rtp.Packet) { + seq := packet.SequenceNumber + if !s.started { + s.packets[seq%s.size] = packet + s.lastAdded = seq + s.started = true + return + } + + diff := seq - s.lastAdded + if diff == 0 { + return + } else if diff < uint16SizeHalf { + for i := s.lastAdded + 1; i != seq; i++ { + s.packets[i%s.size] = nil + } + } + + s.packets[seq%s.size] = packet + s.lastAdded = seq +} + +func (s *sendBuffer) get(seq uint16) *rtp.Packet { + diff := s.lastAdded - seq + if diff >= uint16SizeHalf { + return nil + } + + if diff >= s.size { + return nil + } + + return s.packets[seq%s.size] +} diff --git a/vendor/github.com/pion/interceptor/pkg/report/receiver_interceptor.go b/vendor/github.com/pion/interceptor/pkg/report/receiver_interceptor.go new file mode 100644 index 0000000..5235b99 --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/report/receiver_interceptor.go @@ -0,0 +1,166 @@ +package report + +import ( + "sync" + "time" + + "github.com/pion/interceptor" + "github.com/pion/logging" + "github.com/pion/rtcp" + "github.com/pion/rtp" +) + +// ReceiverInterceptor interceptor generates receiver reports. +type ReceiverInterceptor struct { + interceptor.NoOp + interval time.Duration + now func() time.Time + streams sync.Map + log logging.LeveledLogger + m sync.Mutex + wg sync.WaitGroup + close chan struct{} +} + +// NewReceiverInterceptor returns a new ReceiverInterceptor interceptor. +func NewReceiverInterceptor(opts ...ReceiverOption) (*ReceiverInterceptor, error) { + r := &ReceiverInterceptor{ + interval: 1 * time.Second, + now: time.Now, + log: logging.NewDefaultLoggerFactory().NewLogger("receiver_interceptor"), + close: make(chan struct{}), + } + + for _, opt := range opts { + if err := opt(r); err != nil { + return nil, err + } + } + + return r, nil +} + +func (r *ReceiverInterceptor) isClosed() bool { + select { + case <-r.close: + return true + default: + return false + } +} + +// Close closes the interceptor. +func (r *ReceiverInterceptor) Close() error { + defer r.wg.Wait() + r.m.Lock() + defer r.m.Unlock() + + if !r.isClosed() { + close(r.close) + } + + return nil +} + +// BindRTCPWriter lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method +// will be called once per packet batch. +func (r *ReceiverInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter { + r.m.Lock() + defer r.m.Unlock() + + if r.isClosed() { + return writer + } + + r.wg.Add(1) + + go r.loop(writer) + + return writer +} + +func (r *ReceiverInterceptor) loop(rtcpWriter interceptor.RTCPWriter) { + defer r.wg.Done() + + ticker := time.NewTicker(r.interval) + for { + select { + case <-ticker.C: + now := r.now() + r.streams.Range(func(key, value interface{}) bool { + stream := value.(*receiverStream) + + var pkts []rtcp.Packet + + pkts = append(pkts, stream.generateReport(now)) + + if _, err := rtcpWriter.Write(pkts, interceptor.Attributes{}); err != nil { + r.log.Warnf("failed sending: %+v", err) + } + + return true + }) + + case <-r.close: + return + } + } +} + +// BindRemoteStream lets you modify any incoming RTP packets. It is called once for per RemoteStream. The returned method +// will be called once per rtp packet. +func (r *ReceiverInterceptor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader { + stream := newReceiverStream(info.SSRC, info.ClockRate) + r.streams.Store(info.SSRC, stream) + + return interceptor.RTPReaderFunc(func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) { + i, attr, err := reader.Read(b, a) + if err != nil { + return 0, nil, err + } + + pkt := rtp.Packet{} + if err = pkt.Unmarshal(b[:i]); err != nil { + return 0, nil, err + } + + stream.processRTP(r.now(), &pkt) + + return i, attr, nil + }) +} + +// UnbindLocalStream is called when the Stream is removed. It can be used to clean up any data related to that track. +func (r *ReceiverInterceptor) UnbindLocalStream(info *interceptor.StreamInfo) { + r.streams.Delete(info.SSRC) +} + +// BindRTCPReader lets you modify any incoming RTCP packets. It is called once per sender/receiver, however this might +// change in the future. The returned method will be called once per packet batch. +func (r *ReceiverInterceptor) BindRTCPReader(reader interceptor.RTCPReader) interceptor.RTCPReader { + return interceptor.RTCPReaderFunc(func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) { + i, attr, err := reader.Read(b, a) + if err != nil { + return 0, nil, err + } + + pkts, err := rtcp.Unmarshal(b[:i]) + if err != nil { + return 0, nil, err + } + + for _, pkt := range pkts { + if sr, ok := (pkt).(*rtcp.SenderReport); ok { + value, ok := r.streams.Load(sr.SSRC) + if !ok { + continue + } + + stream := value.(*receiverStream) + stream.processSenderReport(r.now(), sr) + } + } + + return i, attr, nil + }) +} diff --git a/vendor/github.com/pion/interceptor/pkg/report/receiver_option.go b/vendor/github.com/pion/interceptor/pkg/report/receiver_option.go new file mode 100644 index 0000000..0467dc5 --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/report/receiver_option.go @@ -0,0 +1,34 @@ +package report + +import ( + "time" + + "github.com/pion/logging" +) + +// ReceiverOption can be used to configure ReceiverInterceptor. +type ReceiverOption func(r *ReceiverInterceptor) error + +// ReceiverLog sets a logger for the interceptor. +func ReceiverLog(log logging.LeveledLogger) ReceiverOption { + return func(r *ReceiverInterceptor) error { + r.log = log + return nil + } +} + +// ReceiverInterval sets send interval for the interceptor. +func ReceiverInterval(interval time.Duration) ReceiverOption { + return func(r *ReceiverInterceptor) error { + r.interval = interval + return nil + } +} + +// ReceiverNow sets an alternative for the time.Now function. +func ReceiverNow(f func() time.Time) ReceiverOption { + return func(r *ReceiverInterceptor) error { + r.now = f + return nil + } +} diff --git a/vendor/github.com/pion/interceptor/pkg/report/receiver_stream.go b/vendor/github.com/pion/interceptor/pkg/report/receiver_stream.go new file mode 100644 index 0000000..569715d --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/report/receiver_stream.go @@ -0,0 +1,159 @@ +package report + +import ( + "math/rand" + "sync" + "time" + + "github.com/pion/rtcp" + "github.com/pion/rtp" +) + +type receiverStream struct { + ssrc uint32 + receiverSSRC uint32 + clockRate float64 + + m sync.Mutex + size uint16 + packets []uint64 + started bool + seqnumCycles uint16 + lastSeqnum uint16 + lastReportSeqnum uint16 + lastRTPTimeRTP uint32 + lastRTPTimeTime time.Time + jitter float64 + lastSenderReport uint32 + lastSenderReportTime time.Time + totalLost uint32 +} + +func newReceiverStream(ssrc uint32, clockRate uint32) *receiverStream { + receiverSSRC := rand.Uint32() // #nosec + return &receiverStream{ + ssrc: ssrc, + receiverSSRC: receiverSSRC, + clockRate: float64(clockRate), + size: 128, + packets: make([]uint64, 128), + } +} + +func (stream *receiverStream) processRTP(now time.Time, pkt *rtp.Packet) { + stream.m.Lock() + defer stream.m.Unlock() + + if !stream.started { // first frame + stream.started = true + stream.setReceived(pkt.SequenceNumber) + stream.lastSeqnum = pkt.SequenceNumber + stream.lastReportSeqnum = pkt.SequenceNumber - 1 + stream.lastRTPTimeRTP = pkt.Timestamp + stream.lastRTPTimeTime = now + } else { // following frames + stream.setReceived(pkt.SequenceNumber) + + diff := int32(pkt.SequenceNumber) - int32(stream.lastSeqnum) + if diff > 0 || diff < -0x0FFF { + // overflow + if diff < -0x0FFF { + stream.seqnumCycles++ + } + + // set missing packets as missing + for i := stream.lastSeqnum + 1; i != pkt.SequenceNumber; i++ { + stream.delReceived(i) + } + + stream.lastSeqnum = pkt.SequenceNumber + } + + // compute jitter + // https://tools.ietf.org/html/rfc3550#page-39 + D := now.Sub(stream.lastRTPTimeTime).Seconds()*stream.clockRate - + (float64(pkt.Timestamp) - float64(stream.lastRTPTimeRTP)) + if D < 0 { + D = -D + } + stream.jitter += (D - stream.jitter) / 16 + stream.lastRTPTimeRTP = pkt.Timestamp + stream.lastRTPTimeTime = now + } +} + +func (stream *receiverStream) setReceived(seq uint16) { + pos := seq % stream.size + stream.packets[pos/64] |= 1 << (pos % 64) +} + +func (stream *receiverStream) delReceived(seq uint16) { + pos := seq % stream.size + stream.packets[pos/64] &^= 1 << (pos % 64) +} + +func (stream *receiverStream) getReceived(seq uint16) bool { + pos := seq % stream.size + return (stream.packets[pos/64] & (1 << (pos % 64))) != 0 +} + +func (stream *receiverStream) processSenderReport(now time.Time, sr *rtcp.SenderReport) { + stream.m.Lock() + defer stream.m.Unlock() + + stream.lastSenderReport = uint32(sr.NTPTime >> 16) + stream.lastSenderReportTime = now +} + +func (stream *receiverStream) generateReport(now time.Time) *rtcp.ReceiverReport { + stream.m.Lock() + defer stream.m.Unlock() + + totalSinceReport := stream.lastSeqnum - stream.lastReportSeqnum + totalLostSinceReport := func() uint32 { + if stream.lastSeqnum == stream.lastReportSeqnum { + return 0 + } + + ret := uint32(0) + for i := stream.lastReportSeqnum + 1; i != stream.lastSeqnum; i++ { + if !stream.getReceived(i) { + ret++ + } + } + return ret + }() + stream.totalLost += totalLostSinceReport + + // allow up to 24 bits + if totalLostSinceReport > 0xFFFFFF { + totalLostSinceReport = 0xFFFFFF + } + if stream.totalLost > 0xFFFFFF { + stream.totalLost = 0xFFFFFF + } + + r := &rtcp.ReceiverReport{ + SSRC: stream.receiverSSRC, + Reports: []rtcp.ReceptionReport{ + { + SSRC: stream.ssrc, + LastSequenceNumber: uint32(stream.seqnumCycles)<<16 | uint32(stream.lastSeqnum), + LastSenderReport: stream.lastSenderReport, + FractionLost: uint8(float64(totalLostSinceReport*256) / float64(totalSinceReport)), + TotalLost: stream.totalLost, + Delay: func() uint32 { + if stream.lastSenderReportTime.IsZero() { + return 0 + } + return uint32(now.Sub(stream.lastSenderReportTime).Seconds() * 65536) + }(), + Jitter: uint32(stream.jitter), + }, + }, + } + + stream.lastReportSeqnum = stream.lastSeqnum + + return r +} diff --git a/vendor/github.com/pion/interceptor/pkg/report/report.go b/vendor/github.com/pion/interceptor/pkg/report/report.go new file mode 100644 index 0000000..0a3034c --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/report/report.go @@ -0,0 +1,2 @@ +// Package report provides interceptors to implement sending sender and receiver reports. +package report diff --git a/vendor/github.com/pion/interceptor/pkg/report/sender_interceptor.go b/vendor/github.com/pion/interceptor/pkg/report/sender_interceptor.go new file mode 100644 index 0000000..e56fd8c --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/report/sender_interceptor.go @@ -0,0 +1,139 @@ +package report + +import ( + "sync" + "time" + + "github.com/pion/interceptor" + "github.com/pion/logging" + "github.com/pion/rtcp" + "github.com/pion/rtp" +) + +func ntpTime(t time.Time) uint64 { + // seconds since 1st January 1900 + s := (float64(t.UnixNano()) / 1000000000) + 2208988800 + + // higher 32 bits are the integer part, lower 32 bits are the fractional part + integerPart := uint32(s) + fractionalPart := uint32((s - float64(integerPart)) * 0xFFFFFFFF) + return uint64(integerPart)<<32 | uint64(fractionalPart) +} + +// SenderInterceptor interceptor generates sender reports. +type SenderInterceptor struct { + interceptor.NoOp + interval time.Duration + now func() time.Time + streams sync.Map + log logging.LeveledLogger + m sync.Mutex + wg sync.WaitGroup + close chan struct{} +} + +// NewSenderInterceptor returns a new SenderInterceptor interceptor. +func NewSenderInterceptor(opts ...SenderOption) (*SenderInterceptor, error) { + s := &SenderInterceptor{ + interval: 1 * time.Second, + now: time.Now, + log: logging.NewDefaultLoggerFactory().NewLogger("sender_interceptor"), + close: make(chan struct{}), + } + + for _, opt := range opts { + if err := opt(s); err != nil { + return nil, err + } + } + + return s, nil +} + +func (s *SenderInterceptor) isClosed() bool { + select { + case <-s.close: + return true + default: + return false + } +} + +// Close closes the interceptor. +func (s *SenderInterceptor) Close() error { + defer s.wg.Wait() + s.m.Lock() + defer s.m.Unlock() + + if !s.isClosed() { + close(s.close) + } + + return nil +} + +// BindRTCPWriter lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method +// will be called once per packet batch. +func (s *SenderInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter { + s.m.Lock() + defer s.m.Unlock() + + if s.isClosed() { + return writer + } + + s.wg.Add(1) + + go s.loop(writer) + + return writer +} + +func (s *SenderInterceptor) loop(rtcpWriter interceptor.RTCPWriter) { + defer s.wg.Done() + + ticker := time.NewTicker(s.interval) + for { + select { + case <-ticker.C: + now := s.now() + s.streams.Range(func(key, value interface{}) bool { + ssrc := key.(uint32) + stream := value.(*senderStream) + + stream.m.Lock() + defer stream.m.Unlock() + + sr := &rtcp.SenderReport{ + SSRC: ssrc, + NTPTime: ntpTime(now), + RTPTime: stream.lastRTPTimeRTP + uint32(now.Sub(stream.lastRTPTimeTime).Seconds()*stream.clockRate), + PacketCount: stream.packetCount, + OctetCount: stream.octetCount, + } + + if _, err := rtcpWriter.Write([]rtcp.Packet{sr}, interceptor.Attributes{}); err != nil { + s.log.Warnf("failed sending: %+v", err) + } + + return true + }) + + case <-s.close: + return + } + } +} + +// BindLocalStream lets you modify any outgoing RTP packets. It is called once for per LocalStream. The returned method +// will be called once per rtp packet. +func (s *SenderInterceptor) BindLocalStream(info *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter { + stream := newSenderStream(info.ClockRate) + s.streams.Store(info.SSRC, stream) + + return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, a interceptor.Attributes) (int, error) { + stream.processRTP(s.now(), header, payload) + + return writer.Write(header, payload, a) + }) +} diff --git a/vendor/github.com/pion/interceptor/pkg/report/sender_option.go b/vendor/github.com/pion/interceptor/pkg/report/sender_option.go new file mode 100644 index 0000000..4cb161a --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/report/sender_option.go @@ -0,0 +1,34 @@ +package report + +import ( + "time" + + "github.com/pion/logging" +) + +// SenderOption can be used to configure SenderInterceptor. +type SenderOption func(r *SenderInterceptor) error + +// SenderLog sets a logger for the interceptor. +func SenderLog(log logging.LeveledLogger) SenderOption { + return func(r *SenderInterceptor) error { + r.log = log + return nil + } +} + +// SenderInterval sets send interval for the interceptor. +func SenderInterval(interval time.Duration) SenderOption { + return func(r *SenderInterceptor) error { + r.interval = interval + return nil + } +} + +// SenderNow sets an alternative for the time.Now function. +func SenderNow(f func() time.Time) SenderOption { + return func(r *SenderInterceptor) error { + r.now = f + return nil + } +} diff --git a/vendor/github.com/pion/interceptor/pkg/report/sender_stream.go b/vendor/github.com/pion/interceptor/pkg/report/sender_stream.go new file mode 100644 index 0000000..851d70e --- /dev/null +++ b/vendor/github.com/pion/interceptor/pkg/report/sender_stream.go @@ -0,0 +1,37 @@ +package report + +import ( + "sync" + "time" + + "github.com/pion/rtp" +) + +type senderStream struct { + clockRate float64 + m sync.Mutex + + // data from rtp packets + lastRTPTimeRTP uint32 + lastRTPTimeTime time.Time + packetCount uint32 + octetCount uint32 +} + +func newSenderStream(clockRate uint32) *senderStream { + return &senderStream{ + clockRate: float64(clockRate), + } +} + +func (stream *senderStream) processRTP(now time.Time, header *rtp.Header, payload []byte) { + stream.m.Lock() + defer stream.m.Unlock() + + // always update time to minimize errors + stream.lastRTPTimeRTP = header.Timestamp + stream.lastRTPTimeTime = now + + stream.packetCount++ + stream.octetCount += uint32(len(payload)) +} diff --git a/vendor/github.com/pion/interceptor/registry.go b/vendor/github.com/pion/interceptor/registry.go new file mode 100644 index 0000000..d3eed3b --- /dev/null +++ b/vendor/github.com/pion/interceptor/registry.go @@ -0,0 +1,20 @@ +package interceptor + +// Registry is a collector for interceptors. +type Registry struct { + interceptors []Interceptor +} + +// Add adds a new Interceptor to the registry. +func (i *Registry) Add(icpr Interceptor) { + i.interceptors = append(i.interceptors, icpr) +} + +// Build constructs a single Interceptor from a InterceptorRegistry +func (i *Registry) Build() Interceptor { + if len(i.interceptors) == 0 { + return &NoOp{} + } + + return NewChain(i.interceptors) +} diff --git a/vendor/github.com/pion/interceptor/renovate.json b/vendor/github.com/pion/interceptor/renovate.json new file mode 100644 index 0000000..4400fd9 --- /dev/null +++ b/vendor/github.com/pion/interceptor/renovate.json @@ -0,0 +1,15 @@ +{ + "extends": [ + "config:base" + ], + "postUpdateOptions": [ + "gomodTidy" + ], + "commitBody": "Generated by renovateBot", + "packageRules": [ + { + "packagePatterns": ["^golang.org/x/"], + "schedule": ["on the first day of the month"] + } + ] +} diff --git a/vendor/github.com/pion/interceptor/streaminfo.go b/vendor/github.com/pion/interceptor/streaminfo.go new file mode 100644 index 0000000..956fa53 --- /dev/null +++ b/vendor/github.com/pion/interceptor/streaminfo.go @@ -0,0 +1,34 @@ +package interceptor + +// RTPHeaderExtension represents a negotiated RFC5285 RTP header extension. +type RTPHeaderExtension struct { + URI string + ID int +} + +// StreamInfo is the Context passed when a StreamLocal or StreamRemote has been Binded or Unbinded +type StreamInfo struct { + ID string + Attributes Attributes + SSRC uint32 + PayloadType uint8 + RTPHeaderExtensions []RTPHeaderExtension + MimeType string + ClockRate uint32 + Channels uint16 + SDPFmtpLine string + RTCPFeedback []RTCPFeedback +} + +// RTCPFeedback signals the connection to use additional RTCP packet types. +// https://draft.ortc.org/#dom-rtcrtcpfeedback +type RTCPFeedback struct { + // Type is the type of feedback. + // see: https://draft.ortc.org/#dom-rtcrtcpfeedback + // valid: ack, ccm, nack, goog-remb, transport-cc + Type string + + // The parameter value depends on the type. + // For example, type="nack" parameter="pli" will send Picture Loss Indicator packets. + Parameter string +} |