mirror
/
av
mirror of https://bitbucket.org/ausocean/av.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

psi: removed conflict with master

This commit is contained in:
saxon 2019-01-11 13:17:14 +10:30
parent 4e62606622 28d23ad200
commit 9ca7288622
8 changed files with 828 additions and 830 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
revid/senders.go
View File

@ -266,7 +266,7 @@ func newRtmpSender(url string, timeout uint, retries int, log func(lvl int8, msg
var sess *rtmp.Session var sess *rtmp.Session
var err error var err error
for n := 0; n < retries; n++ { for n := 0; n < retries; n++ {
sess = rtmp.NewSession(url, timeout) sess = rtmp.NewSession(url, timeout, log)
err = sess.Open() err = sess.Open()
if err == nil { if err == nil {
break break
@ -312,7 +312,7 @@ func (s *rtmpSender) restart() error {
return err return err
} }
for n := 0; n < s.retries; n++ { for n := 0; n < s.retries; n++ {
s.sess = rtmp.NewSession(s.url, s.timeout) s.sess = rtmp.NewSession(s.url, s.timeout, s.log)
err = s.sess.Open() err = s.sess.Open()
if err == nil { if err == nil {
break break

279
rtmp/packet.go
View File

@ -3,7 +3,7 @@ NAME
packet.go packet.go
DESCRIPTION DESCRIPTION
See Readme.md RTMP packet functionality.
AUTHORS AUTHORS
Saxon Nelson-Milton <saxon@ausocean.org> Saxon Nelson-Milton <saxon@ausocean.org>
@ -36,40 +36,48 @@ package rtmp
import ( import (
"encoding/binary" "encoding/binary"
"log" "io"
) )
// Packet types.
const ( const (
RTMP_PACKET_TYPE_CHUNK_SIZE = 0x01 packetTypeChunkSize = 0x01
RTMP_PACKET_TYPE_BYTES_READ_REPORT = 0x03 packetTypeBytesReadReport = 0x03
RTMP_PACKET_TYPE_CONTROL = 0x04 packetTypeControl = 0x04
RTMP_PACKET_TYPE_SERVER_BW = 0x05 packetTypeServerBW = 0x05
RTMP_PACKET_TYPE_CLIENT_BW = 0x06 packetTypeClientBW = 0x06
RTMP_PACKET_TYPE_AUDIO = 0x08 packetTypeAudio = 0x08
RTMP_PACKET_TYPE_VIDEO = 0x09 packetTypeVideo = 0x09
RTMP_PACKET_TYPE_FLEX_STREAM_SEND = 0x0F packetTypeFlexStreamSend = 0x0F // not implemented
RTMP_PACKET_TYPE_FLEX_SHARED_OBJECT = 0x10 packetTypeFlexSharedObject = 0x10 // not implemented
RTMP_PACKET_TYPE_FLEX_MESSAGE = 0x11 packetTypeFlexMessage = 0x11 // not implemented
RTMP_PACKET_TYPE_INFO = 0x12 packetTypeInfo = 0x12
RTMP_PACKET_TYPE_INVOKE = 0x14 packetTypeInvoke = 0x14
RTMP_PACKET_TYPE_FLASH_VIDEO = 0x16 packetTypeFlashVideo = 0x16 // not implemented
) )
// Header sizes.
const ( const (
RTMP_PACKET_SIZE_LARGE = 0 headerSizeLarge = 0
RTMP_PACKET_SIZE_MEDIUM = 1 headerSizeMedium = 1
RTMP_PACKET_SIZE_SMALL = 2 headerSizeSmall = 2
RTMP_PACKET_SIZE_MINIMUM = 3 headerSizeMinimum = 3
headerSizeAuto = 4
) )
// Special channels.
const ( const (
RTMP_CHANNEL_BYTES_READ = 0x02 chanBytesRead = 0x02
RTMP_CHANNEL_CONTROL = 0x03 chanControl = 0x03
RTMP_CHANNEL_SOURCE = 0x04 chanSource = 0x04
) )
// packetSize defines valid packet sizes. // headerSizes defines header sizes for header types 0, 1, 2 and 3 respectively:
var packetSize = [...]int{12, 8, 4, 1} // 0: full header (12 bytes)
// 1: header without message ID (8 bytes)
// 2: basic header + timestamp (4 byes)
// 3: basic header (chunk type and stream ID) (1 byte)
var headerSizes = [...]int{12, 8, 4, 1}
// packet defines an RTMP packet. // packet defines an RTMP packet.
type packet struct { type packet struct {
@ -90,18 +98,20 @@ type packet struct {
type chunk struct { type chunk struct {
headerSize int32 headerSize int32
data []byte data []byte
header [RTMP_MAX_HEADER_SIZE]byte header [fullHeaderSize]byte
} }
// ToDo: Consider making the following functions into methods. // read reads a packet.
// readPacket reads a packet. func (pkt *packet) read(s *Session) error {
func readPacket(s *Session, pkt *packet) error { var hbuf [fullHeaderSize]byte
var hbuf [RTMP_MAX_HEADER_SIZE]byte
header := hbuf[:] header := hbuf[:]
err := readN(s, header[:1]) _, err := s.read(header[:1])
if err != nil { if err != nil {
log.Println("readPacket: failed to read RTMP packet header!") s.log(DebugLevel, pkg+"failed to read packet header 1st byte", "error", err.Error())
if err == io.EOF {
s.log(WarnLevel, pkg+"EOF error; connection likely terminated")
}
return err return err
} }
pkt.headerType = (header[0] & 0xc0) >> 6 pkt.headerType = (header[0] & 0xc0) >> 6
@ -110,23 +120,22 @@ func readPacket(s *Session, pkt *packet) error {
switch { switch {
case pkt.channel == 0: case pkt.channel == 0:
err = readN(s, header[:1]) _, err = s.read(header[:1])
if err != nil { if err != nil {
log.Println("readPacket: failed to read rtmp packet header 2nd byte.") s.log(DebugLevel, pkg+"failed to read packet header 2nd byte", "error", err.Error())
return err return err
} }
header = header[1:] header = header[1:]
pkt.channel = int32(header[0]) + 64 pkt.channel = int32(header[0]) + 64
case pkt.channel == 1: case pkt.channel == 1:
err = readN(s, header[:2]) _, err = s.read(header[:2])
if err != nil { if err != nil {
log.Println("readPacket: failed to read RTMP packet 3rd byte") s.log(DebugLevel, pkg+"failed to read packet header 3rd byte", "error", err.Error())
return err return err
} }
header = header[2:] header = header[2:]
pkt.channel = int32(binary.BigEndian.Uint16(header[:2])) + 64 pkt.channel = int32(binary.BigEndian.Uint16(header[:2])) + 64
} }
if pkt.channel >= s.channelsAllocatedIn { if pkt.channel >= s.channelsAllocatedIn {
@ -134,49 +143,46 @@ func readPacket(s *Session, pkt *packet) error {
timestamp := append(s.channelTimestamp, make([]int32, 10)...) timestamp := append(s.channelTimestamp, make([]int32, 10)...)
var pkts []*packet var pkts []*packet
if s.vecChannelsIn == nil { if s.channelsIn == nil {
pkts = make([]*packet, n) pkts = make([]*packet, n)
} else { } else {
pkts = append(s.vecChannelsIn[:pkt.channel:pkt.channel], make([]*packet, 10)...) pkts = append(s.channelsIn[:pkt.channel:pkt.channel], make([]*packet, 10)...)
} }
s.channelTimestamp = timestamp s.channelTimestamp = timestamp
s.vecChannelsIn = pkts s.channelsIn = pkts
for i := int(s.channelsAllocatedIn); i < len(s.channelTimestamp); i++ { for i := int(s.channelsAllocatedIn); i < len(s.channelTimestamp); i++ {
s.channelTimestamp[i] = 0 s.channelTimestamp[i] = 0
} }
for i := int(s.channelsAllocatedIn); i < int(n); i++ { for i := int(s.channelsAllocatedIn); i < int(n); i++ {
s.vecChannelsIn[i] = nil s.channelsIn[i] = nil
} }
s.channelsAllocatedIn = n s.channelsAllocatedIn = n
} }
size := packetSize[pkt.headerType] size := headerSizes[pkt.headerType]
switch { switch {
case size == RTMP_LARGE_HEADER_SIZE: case size == fullHeaderSize:
pkt.hasAbsTimestamp = true pkt.hasAbsTimestamp = true
case size < RTMP_LARGE_HEADER_SIZE: case size < fullHeaderSize:
if s.vecChannelsIn[pkt.channel] != nil { if s.channelsIn[pkt.channel] != nil {
*pkt = *(s.vecChannelsIn[pkt.channel]) *pkt = *(s.channelsIn[pkt.channel])
} }
} }
size-- size--
if size > 0 { if size > 0 {
err = readN(s, header[:size]) _, err = s.read(header[:size])
if err != nil { if err != nil {
log.Println("readPacket: failed to read rtmp packet heades.") s.log(DebugLevel, pkg+"failed to read packet header", "error", err.Error())
return err return err
} }
} }
hSize := len(hbuf) - len(header) + size hSize := len(hbuf) - len(header) + size
if size >= 3 { if size >= 3 {
pkt.timestamp = C_AMF_DecodeInt24(header[:3]) pkt.timestamp = C_AMF_DecodeInt24(header[:3])
if size >= 6 { if size >= 6 {
pkt.bodySize = C_AMF_DecodeInt24(header[3:6]) pkt.bodySize = C_AMF_DecodeInt24(header[3:6])
pkt.bytesRead = 0 pkt.bytesRead = 0
@ -193,9 +199,9 @@ func readPacket(s *Session, pkt *packet) error {
extendedTimestamp := pkt.timestamp == 0xffffff extendedTimestamp := pkt.timestamp == 0xffffff
if extendedTimestamp { if extendedTimestamp {
err = readN(s, header[size:size+4]) _, err = s.read(header[size : size+4])
if err != nil { if err != nil {
log.Println("readPacket: Failed to read extended timestamp") s.log(DebugLevel, pkg+"failed to read extended timestamp", "error", err.Error())
return err return err
} }
// TODO: port this // TODO: port this
@ -204,7 +210,7 @@ func readPacket(s *Session, pkt *packet) error {
} }
if pkt.bodySize > 0 && pkt.body == nil { if pkt.bodySize > 0 && pkt.body == nil {
resizePacket(pkt, pkt.bodySize, (hbuf[0]&0xc0)>>6) pkt.resize(pkt.bodySize, (hbuf[0]&0xc0)>>6)
} }
toRead := int32(pkt.bodySize - pkt.bytesRead) toRead := int32(pkt.bodySize - pkt.bytesRead)
@ -215,27 +221,28 @@ func readPacket(s *Session, pkt *packet) error {
} }
if pkt.chunk != nil { if pkt.chunk != nil {
panic("non-nil chunk")
pkt.chunk.headerSize = int32(hSize) pkt.chunk.headerSize = int32(hSize)
copy(pkt.chunk.header[:], hbuf[:hSize]) copy(pkt.chunk.header[:], hbuf[:hSize])
pkt.chunk.data = pkt.body[pkt.bytesRead : pkt.bytesRead+uint32(chunkSize)] pkt.chunk.data = pkt.body[pkt.bytesRead : pkt.bytesRead+uint32(chunkSize)]
} }
err = readN(s, pkt.body[pkt.bytesRead:][:chunkSize]) _, err = s.read(pkt.body[pkt.bytesRead:][:chunkSize])
if err != nil { if err != nil {
log.Println("readPacket: failed to read RTMP packet body") s.log(DebugLevel, pkg+"failed to read packet body", "error", err.Error())
return err return err
} }
pkt.bytesRead += uint32(chunkSize) pkt.bytesRead += uint32(chunkSize)
// keep the packet as ref for other packets on this channel // keep the packet as ref for other packets on this channel
if s.vecChannelsIn[pkt.channel] == nil { if s.channelsIn[pkt.channel] == nil {
s.vecChannelsIn[pkt.channel] = &packet{} s.channelsIn[pkt.channel] = &packet{}
} }
*(s.vecChannelsIn[pkt.channel]) = *pkt *(s.channelsIn[pkt.channel]) = *pkt
if extendedTimestamp { if extendedTimestamp {
s.vecChannelsIn[pkt.channel].timestamp = 0xffffff s.channelsIn[pkt.channel].timestamp = 0xffffff
} }
if pkt.bytesRead != pkt.bodySize { if pkt.bytesRead != pkt.bodySize {
@ -248,77 +255,90 @@ func readPacket(s *Session, pkt *packet) error {
} }
s.channelTimestamp[pkt.channel] = int32(pkt.timestamp) s.channelTimestamp[pkt.channel] = int32(pkt.timestamp)
s.vecChannelsIn[pkt.channel].body = nil s.channelsIn[pkt.channel].body = nil
s.vecChannelsIn[pkt.channel].bytesRead = 0 s.channelsIn[pkt.channel].bytesRead = 0
s.vecChannelsIn[pkt.channel].hasAbsTimestamp = false s.channelsIn[pkt.channel].hasAbsTimestamp = false
return nil return nil
} }
// resizePacket adjust the packet's storage to accommodate a body of the given size. // resize adjusts the packet's storage to accommodate a body of the given size and header type.
func resizePacket(pkt *packet, size uint32, ht uint8) { func (pkt *packet) resize(size uint32, ht uint8) {
buf := make([]byte, RTMP_MAX_HEADER_SIZE+size) buf := make([]byte, fullHeaderSize+size)
pkt.headerType = ht
pkt.header = buf pkt.header = buf
pkt.body = buf[RTMP_MAX_HEADER_SIZE:] pkt.body = buf[fullHeaderSize:]
if ht != headerSizeAuto {
pkt.headerType = ht
return
}
switch pkt.packetType {
case packetTypeVideo, packetTypeAudio:
if pkt.timestamp == 0 {
pkt.headerType = headerSizeLarge
} else {
pkt.headerType = headerSizeMedium
}
case packetTypeInfo:
pkt.headerType = headerSizeLarge
pkt.bodySize += 16
default:
pkt.headerType = headerSizeMedium
}
} }
// sendPacket sends a packet. // write sends a packet.
func sendPacket(s *Session, pkt *packet, queue bool) error { // When queue is true, we expect a response to this request and cache the method on s.methodCalls.
var prevPkt *packet func (pkt *packet) write(s *Session, queue bool) error {
var last int if pkt.body == nil {
return errInvalidBody
}
if pkt.channel >= s.channelsAllocatedOut { if pkt.channel >= s.channelsAllocatedOut {
s.log(DebugLevel, pkg+"growing channelsOut", "channel", pkt.channel)
n := int(pkt.channel + 10) n := int(pkt.channel + 10)
var pkts []*packet var pkts []*packet
if s.vecChannelsOut == nil { if s.channelsOut == nil {
pkts = make([]*packet, n) pkts = make([]*packet, n)
} else { } else {
pkts = append(s.vecChannelsOut[:pkt.channel:pkt.channel], make([]*packet, 10)...) pkts = append(s.channelsOut[:pkt.channel:pkt.channel], make([]*packet, 10)...)
} }
s.vecChannelsOut = pkts s.channelsOut = pkts
for i := int(s.channelsAllocatedOut); i < n; i++ { for i := int(s.channelsAllocatedOut); i < n; i++ {
s.vecChannelsOut[i] = nil s.channelsOut[i] = nil
} }
s.channelsAllocatedOut = int32(n) s.channelsAllocatedOut = int32(n)
} }
prevPkt = s.vecChannelsOut[pkt.channel]
if prevPkt != nil && pkt.headerType != RTMP_PACKET_SIZE_LARGE { prevPkt := s.channelsOut[pkt.channel]
var last int
if prevPkt != nil && pkt.headerType != headerSizeLarge {
// compress a bit by using the prev packet's attributes // compress a bit by using the prev packet's attributes
if prevPkt.bodySize == pkt.bodySize && prevPkt.packetType == pkt.packetType && pkt.headerType == RTMP_PACKET_SIZE_MEDIUM { if prevPkt.bodySize == pkt.bodySize && prevPkt.packetType == pkt.packetType && pkt.headerType == headerSizeMedium {
pkt.headerType = RTMP_PACKET_SIZE_SMALL pkt.headerType = headerSizeSmall
} }
if prevPkt.timestamp == pkt.timestamp && pkt.headerType == RTMP_PACKET_SIZE_SMALL { if prevPkt.timestamp == pkt.timestamp && pkt.headerType == headerSizeSmall {
pkt.headerType = RTMP_PACKET_SIZE_MINIMUM pkt.headerType = headerSizeMinimum
} }
last = int(prevPkt.timestamp) last = int(prevPkt.timestamp)
} }
if pkt.headerType > 3 { if pkt.headerType > 3 {
log.Printf("Sanity failed! trying to send header of type: 0x%02x.", s.log(WarnLevel, pkg+"unexpected header type", "type", pkt.headerType)
pkt.headerType)
return errInvalidHeader return errInvalidHeader
} }
var headBytes []byte // The complete packet starts from headerSize _before_ the start the body.
var origIdx int // origIdx is the original offset, which will be 0 for a full (12-byte) header or 11 for a minimum (1-byte) header.
if pkt.body != nil { headBytes := pkt.header
// Span from -packetsize for the type to the start of the body. hSize := headerSizes[pkt.headerType]
headBytes = pkt.header origIdx := fullHeaderSize - hSize
origIdx = RTMP_MAX_HEADER_SIZE - packetSize[pkt.headerType]
} else {
// Allocate a new header and allow 6 bytes of movement backward.
var hbuf [RTMP_MAX_HEADER_SIZE]byte
headBytes = hbuf[:]
origIdx = 6
}
var cSize int // adjust 1 or 2 bytes for the channel
cSize := 0
switch { switch {
case pkt.channel > 319: case pkt.channel > 319:
cSize = 2 cSize = 2
@ -326,12 +346,12 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
cSize = 1 cSize = 1
} }
hSize := packetSize[pkt.headerType]
if cSize != 0 { if cSize != 0 {
origIdx -= cSize origIdx -= cSize
hSize += cSize hSize += cSize
} }
// adjust 4 bytes for the timestamp
var ts uint32 var ts uint32
if prevPkt != nil { if prevPkt != nil {
ts = uint32(int(pkt.timestamp) - last) ts = uint32(int(pkt.timestamp) - last)
@ -339,7 +359,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
if ts >= 0xffffff { if ts >= 0xffffff {
origIdx -= 4 origIdx -= 4
hSize += 4 hSize += 4
log.Printf("Larger timestamp than 24-bit: 0x%v", ts) s.log(DebugLevel, pkg+"larger timestamp than 24 bits", "timestamp", ts)
} }
headerIdx := origIdx headerIdx := origIdx
@ -367,7 +387,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
} }
} }
if packetSize[pkt.headerType] > 1 { if headerSizes[pkt.headerType] > 1 {
res := ts res := ts
if ts > 0xffffff { if ts > 0xffffff {
res = 0xffffff res = 0xffffff
@ -376,16 +396,16 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
headerIdx += 3 // 24bits headerIdx += 3 // 24bits
} }
if packetSize[pkt.headerType] > 4 { if headerSizes[pkt.headerType] > 4 {
C_AMF_EncodeInt24(headBytes[headerIdx:], int32(pkt.bodySize)) C_AMF_EncodeInt24(headBytes[headerIdx:], int32(pkt.bodySize))
headerIdx += 3 // 24bits headerIdx += 3 // 24bits
headBytes[headerIdx] = pkt.packetType headBytes[headerIdx] = pkt.packetType
headerIdx++ headerIdx++
} }
if packetSize[pkt.headerType] > 8 { if headerSizes[pkt.headerType] > 8 {
n := int(encodeInt32LE(headBytes[headerIdx:headerIdx+4], pkt.info)) binary.LittleEndian.PutUint32(headBytes[headerIdx:headerIdx+4], uint32(pkt.info))
headerIdx += n headerIdx += 4 // 32bits
} }
if ts >= 0xffffff { if ts >= 0xffffff {
@ -396,39 +416,43 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
size := int(pkt.bodySize) size := int(pkt.bodySize)
chunkSize := int(s.outChunkSize) chunkSize := int(s.outChunkSize)
if debugMode { if s.deferred == nil {
log.Printf("sendPacket: %v->%v, size=%v", s.link.conn.LocalAddr(), s.link.conn.RemoteAddr(), size) // Defer sending small audio packets (at most once).
} if pkt.packetType == packetTypeAudio && size < chunkSize {
s.deferred = headBytes[origIdx:][:size+hSize]
// Send the previously deferred packet if combining it with the next packet would exceed the chunk size. s.log(DebugLevel, pkg+"deferred sending packet", "size", size, "la", s.link.conn.LocalAddr(), "ra", s.link.conn.RemoteAddr())
if s.defered != nil && len(s.defered)+size+hSize > chunkSize { return nil
err := writeN(s, s.defered) }
if err != nil { } else {
return err // Send previously deferrd packet if combining it with the next one would exceed the chunk size.
if len(s.deferred)+size+hSize > chunkSize {
s.log(DebugLevel, pkg+"sending deferred packet separately", "size", len(s.deferred))
_, err := s.write(s.deferred)
if err != nil {
return err
}
s.deferred = nil
} }
s.defered = nil
} }
// TODO(kortschak): Rewrite this horrific peice of premature optimisation. // TODO(kortschak): Rewrite this horrific peice of premature optimisation.
// NB: RTMP wants packets in chunks which are 128 bytes by default, but the server may request a different size. // NB: RTMP wants packets in chunks which are 128 bytes by default, but the server may request a different size.
s.log(DebugLevel, pkg+"sending packet", "la", s.link.conn.LocalAddr(), "ra", s.link.conn.RemoteAddr(), "size", size)
for size+hSize != 0 { for size+hSize != 0 {
if s.defered == nil && pkt.packetType == RTMP_PACKET_TYPE_AUDIO && size < chunkSize {
s.defered = headBytes[origIdx:][:size+hSize]
break
}
if chunkSize > size { if chunkSize > size {
chunkSize = size chunkSize = size
} }
bytes := headBytes[origIdx:][:chunkSize+hSize] bytes := headBytes[origIdx:][:chunkSize+hSize]
if s.defered != nil { if s.deferred != nil {
// Prepend the previously deferred packet and write it with the current one. // Prepend the previously deferred packet and write it with the current one.
bytes = append(s.defered, bytes...) s.log(DebugLevel, pkg+"combining deferred packet", "size", len(s.deferred))
bytes = append(s.deferred, bytes...)
} }
err := writeN(s, bytes) _, err := s.write(bytes)
if err != nil { if err != nil {
return err return err
} }
s.defered = nil s.deferred = nil
size -= chunkSize size -= chunkSize
origIdx += chunkSize + hSize origIdx += chunkSize + hSize
@ -461,13 +485,10 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
} }
// We invoked a remote method // We invoked a remote method
if pkt.packetType == RTMP_PACKET_TYPE_INVOKE { if pkt.packetType == packetTypeInvoke {
buf := pkt.body[1:] buf := pkt.body[1:]
meth := C_AMF_DecodeString(buf) meth := C_AMF_DecodeString(buf)
s.log(DebugLevel, pkg+"invoking method "+meth)
if debugMode {
log.Printf("invoking %v", meth)
}
// keep it in call queue till result arrives // keep it in call queue till result arrives
if queue { if queue {
buf = buf[3+len(meth):] buf = buf[3+len(meth):]
@ -476,10 +497,10 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
} }
} }
if s.vecChannelsOut[pkt.channel] == nil { if s.channelsOut[pkt.channel] == nil {
s.vecChannelsOut[pkt.channel] = &packet{} s.channelsOut[pkt.channel] = &packet{}
} }
*(s.vecChannelsOut[pkt.channel]) = *pkt *(s.channelsOut[pkt.channel]) = *pkt
return nil return nil
} }

21
rtmp/parseurl.go
View File

@ -8,9 +8,10 @@ DESCRIPTION
AUTHOR AUTHOR
Dan Kortschak <dan@ausocean.org> Dan Kortschak <dan@ausocean.org>
Saxon Nelson-Milton <saxon@ausocean.org> Saxon Nelson-Milton <saxon@ausocean.org>
Alan Noble <alan@ausocean.org>
LICENSE LICENSE
parseurl.go is Copyright (C) 2017-2018 the Australian Ocean Lab (AusOcean) parseurl.go is Copyright (C) 2017-2019 the Australian Ocean Lab (AusOcean)
It is free software: you can redistribute it and/or modify them It is free software: you can redistribute it and/or modify them
under the terms of the GNU General Public License as published by the under the terms of the GNU General Public License as published by the
@ -33,7 +34,6 @@ LICENSE
package rtmp package rtmp
import ( import (
"log"
"net/url" "net/url"
"path" "path"
"strconv" "strconv"
@ -41,30 +41,29 @@ import (
) )
// parseURL parses an RTMP URL (ok, technically it is lexing). // parseURL parses an RTMP URL (ok, technically it is lexing).
//
func parseURL(addr string) (protocol int32, host string, port uint16, app, playpath string, err error) { func parseURL(addr string) (protocol int32, host string, port uint16, app, playpath string, err error) {
u, err := url.Parse(addr) u, err := url.Parse(addr)
if err != nil { if err != nil {
log.Printf("failed to parse addr: %v", err)
return protocol, host, port, app, playpath, err return protocol, host, port, app, playpath, err
} }
switch u.Scheme { switch u.Scheme {
case "rtmp": case "rtmp":
protocol = RTMP_PROTOCOL_RTMP protocol = protoRTMP
case "rtmpt": case "rtmpt":
protocol = RTMP_PROTOCOL_RTMPT protocol = protoRTMPT
case "rtmps": case "rtmps":
protocol = RTMP_PROTOCOL_RTMPS protocol = protoRTMPS
case "rtmpe": case "rtmpe":
protocol = RTMP_PROTOCOL_RTMPE protocol = protoRTMPE
case "rtmfp": case "rtmfp":
protocol = RTMP_PROTOCOL_RTMFP protocol = protoRTMFP
case "rtmpte": case "rtmpte":
protocol = RTMP_PROTOCOL_RTMPTE protocol = protoRTMPTE
case "rtmpts": case "rtmpts":
protocol = RTMP_PROTOCOL_RTMPTS protocol = protoRTMPTS
default: default:
log.Printf("unknown scheme: %q", u.Scheme)
return protocol, host, port, app, playpath, errUnknownScheme return protocol, host, port, app, playpath, errUnknownScheme
} }

710
rtmp/rtmp.go
View File

File diff suppressed because it is too large Load Diff

119
rtmp/rtmp_headers.go
View File

@ -1,119 +0,0 @@
/*
NAME
rtmp_headers.go
DESCRIPTION
See Readme.md
AUTHORS
Saxon Nelson-Milton <saxon@ausocean.org>
Dan Kortschak <dan@ausocean.org>
Alan Noble <alan@ausocean.org>
LICENSE
rtmp_headers.go is Copyright (C) 2017-2019 the Australian Ocean Lab (AusOcean)
It is free software: you can redistribute it and/or modify them
under the terms of the GNU General Public License as published by the
Free Software Foundation, either version 3 of the License, or (at your
option) any later version.
It is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
for more details.
You should have received a copy of the GNU General Public License
along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
Derived from librtmp under the GNU Lesser General Public License 2.1
Copyright (C) 2005-2008 Team XBMC http://www.xbmc.org
Copyright (C) 2008-2009 Andrej Stepanchuk
Copyright (C) 2009-2010 Howard Chu
*/
package rtmp
import (
"net"
)
const (
RTMPT_OPEN = iota
RTMPT_SEND
RTMPT_IDLE
RTMPT_CLOSE
)
const (
RTMP_READ_HEADER = 0x01
RTMP_READ_RESUME = 0x02
RTMP_READ_NO_IGNORE = 0x04
RTMP_READ_GOTKF = 0x08
RTMP_READ_GOTFLVK = 0x10
RTMP_READ_SEEKING = 0x20
RTMP_READ_COMPLETE = -3
RTMP_READ_ERROR = -2
RTMP_READ_EOF = -1
RTMP_READ_IGNORE = 0
)
const (
RTMP_LF_AUTH = 0x0001 /* using auth param */
RTMP_LF_LIVE = 0x0002 /* stream is live */
RTMP_LF_SWFV = 0x0004 /* do SWF verification */
RTMP_LF_PLST = 0x0008 /* send playlist before play */
RTMP_LF_BUFX = 0x0010 /* toggle stream on BufferEmpty msg */
RTMP_LF_FTCU = 0x0020 /* free tcUrl on close */
RTMP_LF_FAPU = 0x0040 /* free app on close */
)
const (
RTMP_FEATURE_HTTP = 0x01
RTMP_FEATURE_ENC = 0x02
RTMP_FEATURE_SSL = 0x04
RTMP_FEATURE_MFP = 0x08 /* not yet supported */
RTMP_FEATURE_WRITE = 0x10 /* publish, not play */
RTMP_FEATURE_HTTP2 = 0x20 /* server-side rtmpt */
)
const (
RTMP_PROTOCOL_RTMP = 0
RTMP_PROTOCOL_RTMPE = RTMP_FEATURE_ENC
RTMP_PROTOCOL_RTMPT = RTMP_FEATURE_HTTP
RTMP_PROTOCOL_RTMPS = RTMP_FEATURE_SSL
RTMP_PROTOCOL_RTMPTE = (RTMP_FEATURE_HTTP | RTMP_FEATURE_ENC)
RTMP_PROTOCOL_RTMPTS = (RTMP_FEATURE_HTTP | RTMP_FEATURE_SSL)
RTMP_PROTOCOL_RTMFP = RTMP_FEATURE_MFP
)
const (
RTMP_DEFAULT_CHUNKSIZE = 128
RTMP_BUFFER_CACHE_SIZE = (16 * 1024)
RTMP_SIG_SIZE = 1536
RTMP_LARGE_HEADER_SIZE = 12
RTMP_MAX_HEADER_SIZE = 18
)
type link struct {
host string
playpath string
tcUrl string
swfUrl string
pageUrl string
app string
auth string
flashVer string
token string
extras C_AMFObject
lFlags int32
swfAge int32
protocol int32
timeout uint
port uint16
conn *net.TCPConn
}
type method struct {
name string
num int32
}

232
rtmp/rtmp_test.go Normal file
View File

@ -0,0 +1,232 @@
/*
NAME
rtmp_test.go
DESCRIPTION
RTMP tests
AUTHORS
Saxon Nelson-Milton <saxon@ausocean.org>
Dan Kortschak <dan@ausocean.org>
Alan Noble <alan@ausocean.org>
LICENSE
rtmp_test.go is Copyright (C) 2017-2019 the Australian Ocean Lab (AusOcean)
It is free software: you can redistribute it and/or modify them
under the terms of the GNU General Public License as published by the
Free Software Foundation, either version 3 of the License, or (at your
option) any later version.
It is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
for more details.
You should have received a copy of the GNU General Public License
along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
*/
package rtmp
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"runtime"
"testing"
"time"
"bitbucket.org/ausocean/av/stream/flv"
"bitbucket.org/ausocean/av/stream/lex"
)
const (
rtmpProtocol = "rtmp"
testHost = "a.rtmp.youtube.com"
testApp = "live2"
testBaseURL = rtmpProtocol + "://" + testHost + "/" + testApp + "/"
testTimeout = 30
testDataDir = "../../test/test-data/av/input"
)
// testVerbosity controls the amount of output.
// NB: This is not the log level, which is DebugLevel.
// 0: suppress logging completely
// 1: log messages only
// 2: log messages with errors, if any
var testVerbosity = 1
// testKey is the YouTube RTMP key required for YouTube streaming (RTMP_TEST_KEY env var).
// NB: don't share your key with others.
var testKey string
// testFile is the test video file (RTMP_TEST_FILE env var).
// betterInput.h264 is a good one to use.
var testFile string
// testLog is a bare bones logger that logs to stdout, and exits upon either an error or fatal error.
func testLog(level int8, msg string, params ...interface{}) {
logLevels := [...]string{"Debug", "Info", "Warn", "Error", "", "", "Fatal"}
if testVerbosity == 0 {
return
}
if level < -1 || level > 5 {
panic("Invalid log level")
}
switch testVerbosity {
case 0:
// silence is golden
case 1:
fmt.Printf("%s: %s\n", logLevels[level+1], msg)
case 2:
// extract the first param if it is one we care about, otherwise just print the message
if len(params) >= 2 {
switch params[0].(string) {
case "error":
fmt.Printf("%s: %s, error=%v\n", logLevels[level+1], msg, params[1].(string))
case "size":
fmt.Printf("%s: %s, size=%d\n", logLevels[level+1], msg, params[1].(int))
default:
fmt.Printf("%s: %s\n", logLevels[level+1], msg)
}
} else {
fmt.Printf("%s: %s\n", logLevels[level+1], msg)
}
}
if level >= 4 {
// Error or Fatal
buf := make([]byte, 1<<16)
size := runtime.Stack(buf, true)
fmt.Printf("%s\n", string(buf[:size]))
os.Exit(1)
}
}
// TestKey tests that the RTMP_TEST_KEY environment variable is present
func TestKey(t *testing.T) {
testLog(0, "TestKey")
testKey = os.Getenv("RTMP_TEST_KEY")
if testKey == "" {
msg := "RTMP_TEST_KEY environment variable not defined"
testLog(0, msg)
t.Skip(msg)
}
testLog(0, "Testing against URL "+testBaseURL+testKey)
}
// TestSetupURL tests URL parsing.
func TestSetupURL(t *testing.T) {
testLog(0, "TestSetupURL")
// test with just the base URL
s := NewSession(testBaseURL, testTimeout, testLog)
if s.url != testBaseURL && s.link.timeout != testTimeout {
t.Errorf("NewSession failed")
}
err := setupURL(s)
if err != nil {
t.Errorf("setupURL(testBaseURL) failed with error: %v", err)
}
// test the parts are as expected
if rtmpProtocolStrings[s.link.protocol] != rtmpProtocol {
t.Errorf("setupURL returned wrong protocol: %v", s.link.protocol)
}
if s.link.host != testHost {
t.Errorf("setupURL returned wrong host: %v", s.link.host)
}
if s.link.app != testApp {
t.Errorf("setupURL returned wrong app: %v", s.link.app)
}
}
// TestOpenClose tests opening an closing an RTMP connection.
func TestOpenClose(t *testing.T) {
testLog(0, "TestOpenClose")
if testKey == "" {
t.Skip("Skipping TestOpenClose since no RTMP_TEST_KEY")
}
s := NewSession(testBaseURL+testKey, testTimeout, testLog)
err := s.Open()
if err != nil {
t.Errorf("Open failed with error: %v", err)
return
}
err = s.Close()
if err != nil {
t.Errorf("Close failed with error: %v", err)
return
}
}
// TestFromFrame tests streaming from a single H.264 frame which is repeated.
func TestFromFrame(t *testing.T) {
testLog(0, "TestFromFrame")
if testKey == "" {
t.Skip("Skipping TestFromFrame since no RTMP_TEST_KEY")
}
s := NewSession(testBaseURL+testKey, testTimeout, testLog)
err := s.Open()
if err != nil {
t.Errorf("Session.Open failed with error: %v", err)
}
b, err := ioutil.ReadFile(filepath.Join(testDataDir, "AusOcean_logo_1080p.h264"))
if err != nil {
t.Errorf("ReadFile failed with error: %v", err)
}
// Pass RTMP session, true for audio, true for video, and 25 FPS
// ToDo: fix this. Although we can encode the file and YouTube
// doesn't complain, YouTube doesn't play it (even when we
// send 1 minute's worth).
flvEncoder := flv.NewEncoder(s, true, true, 25)
for i := 0; i < 25; i++ {
err := flvEncoder.Encode(b)
if err != nil {
t.Errorf("Encoding failed with error: %v", err)
}
time.Sleep(time.Millisecond / 25) // rate limit to 1/25s
}
err = s.Close()
if err != nil {
t.Errorf("Session.Close failed with error: %v", err)
}
}
// TestFromFile tests streaming from an video file comprising raw H.264.
// The test file is supplied via the RTMP_TEST_FILE environment variable.
func TestFromFile(t *testing.T) {
testLog(0, "TestFromFile")
testFile := os.Getenv("RTMP_TEST_FILE")
if testFile == "" {
t.Skip("Skipping TestFromFile since no RTMP_TEST_FILE")
}
if testKey == "" {
t.Skip("Skipping TestFromFile since no RTMP_TEST_KEY")
}
s := NewSession(testBaseURL+testKey, testTimeout, testLog)
err := s.Open()
if err != nil {
t.Errorf("Session.Open failed with error: %v", err)
}
f, err := os.Open(testFile)
if err != nil {
t.Errorf("Open failed with error: %v", err)
}
defer f.Close()
// Pass RTMP session, true for audio, true for video, and 25 FPS
flvEncoder := flv.NewEncoder(s, true, true, 25)
err = lex.H264(flvEncoder, f, time.Second/time.Duration(25))
if err != nil {
t.Errorf("Lexing and encoding failed with error: %v", err)
}
err = s.Close()
if err != nil {
t.Errorf("Session.Close failed with error: %v", err)
}
}

190
rtmp/session.go
View File

@ -3,7 +3,7 @@ NAME
session.go session.go
DESCRIPTION DESCRIPTION
See Readme.md RTMP session functionality.
AUTHORS AUTHORS
Saxon Nelson-Milton <saxon@ausocean.org> Saxon Nelson-Milton <saxon@ausocean.org>
@ -34,16 +34,17 @@ LICENSE
package rtmp package rtmp
import ( import (
"errors" "io"
"net"
"time"
) )
// Session holds the state for an RTMP session. // Session holds the state for an RTMP session.
type Session struct { type Session struct {
url string url string
timeout uint
inChunkSize int32 inChunkSize int32
outChunkSize int32 outChunkSize int32
bwCheckCounter int32 checkCounter int32
nBytesIn int32 nBytesIn int32
nBytesInSent int32 nBytesInSent int32
streamID int32 streamID int32
@ -56,96 +57,116 @@ type Session struct {
methodCalls []method methodCalls []method
channelsAllocatedIn int32 channelsAllocatedIn int32
channelsAllocatedOut int32 channelsAllocatedOut int32
vecChannelsIn []*packet channelsIn []*packet
vecChannelsOut []*packet channelsOut []*packet
channelTimestamp []int32 channelTimestamp []int32
audioCodecs float64 audioCodecs float64
videoCodecs float64 videoCodecs float64
encoding float64 encoding float64
defered []byte deferred []byte
link link link link
log Log
} }
// link represents RTMP URL and connection information.
type link struct {
host string
playpath string
tcUrl string
swfUrl string
pageUrl string
app string
auth string
flashVer string
token string
extras C_AMFObject
flags int32
swfAge int32
protocol int32
timeout uint
port uint16
conn *net.TCPConn
}
// method represents an RTMP method.
type method struct {
name string
num int32
}
// Log defines the RTMP logging function.
type Log func(level int8, message string, params ...interface{})
// Log levels used by Log.
const (
DebugLevel int8 = -1
InfoLevel int8 = 0
WarnLevel int8 = 1
ErrorLevel int8 = 2
FatalLevel int8 = 5
)
// NewSession returns a new Session. // NewSession returns a new Session.
func NewSession(url string, connectTimeout uint) *Session { func NewSession(url string, timeout uint, log Log) *Session {
return &Session{ return &Session{
url: url, url: url,
timeout: connectTimeout, inChunkSize: 128,
outChunkSize: 128,
clientBW: 2500000,
clientBW2: 2,
serverBW: 2500000,
audioCodecs: 3191.0,
videoCodecs: 252.0,
log: log,
link: link{
timeout: timeout,
swfAge: 30,
},
} }
} }
// Open establishes an rtmp connection with the url passed into the constructor. // Open establishes an rtmp connection with the url passed into the constructor.
func (s *Session) Open() error { func (s *Session) Open() error {
s.log(DebugLevel, pkg+"Session.Open")
if s.isConnected() { if s.isConnected() {
return errors.New("rtmp: attempt to start already running session") return errConnected
} }
err := s.start() err := setupURL(s)
if err != nil { if err != nil {
return err return err
} }
return nil
}
// start does the heavylifting for Open().
func (s *Session) start() error {
s.init()
err := setupURL(s, s.url)
if err != nil {
s.close()
return err
}
s.enableWrite() s.enableWrite()
err = connect(s) err = connect(s)
if err != nil { if err != nil {
s.close() s.Close()
return err return err
} }
err = connectStream(s) err = connectStream(s)
if err != nil { if err != nil {
s.close() s.Close()
return err return err
} }
return nil return nil
} }
// init initializes various RTMP defauls. // Close terminates the rtmp connection.
// ToDo: define consts for the magic numbers. // NB: Close is idempotent and the session value is cleared completely.
func (s *Session) init() {
s.inChunkSize = RTMP_DEFAULT_CHUNKSIZE
s.outChunkSize = RTMP_DEFAULT_CHUNKSIZE
s.clientBW = 2500000
s.clientBW2 = 2
s.serverBW = 2500000
s.audioCodecs = 3191.0
s.videoCodecs = 252.0
s.link.timeout = s.timeout
s.link.swfAge = 30
}
// Close terminates the rtmp connection,
func (s *Session) Close() error { func (s *Session) Close() error {
s.log(DebugLevel, pkg+"Session.Close")
if !s.isConnected() { if !s.isConnected() {
return errNotConnected return errNotConnected
} }
s.close() if s.streamID > 0 {
return nil if s.link.protocol&featureWrite != 0 {
} sendFCUnpublish(s)
// close does the heavylifting for Close().
// Any errors are ignored as it is often called in response to an earlier error.
func (s *Session) close() {
if s.isConnected() {
if s.streamID > 0 {
if s.link.protocol&RTMP_FEATURE_WRITE != 0 {
sendFCUnpublish(s)
}
sendDeleteStream(s, float64(s.streamID))
} }
s.link.conn.Close() sendDeleteStream(s, float64(s.streamID))
} }
s.link.conn.Close()
*s = Session{} *s = Session{}
return nil
} }
// Write writes a frame (flv tag) to the rtmp connection. // Write writes a frame (flv tag) to the rtmp connection.
@ -153,13 +174,70 @@ func (s *Session) Write(data []byte) (int, error) {
if !s.isConnected() { if !s.isConnected() {
return 0, errNotConnected return 0, errNotConnected
} }
err := s.write(data) if len(data) < minDataSize {
return 0, errTinyPacket
}
if data[0] == packetTypeInfo || (data[0] == 'F' && data[1] == 'L' && data[2] == 'V') {
return 0, errUnimplemented
}
pkt := packet{
packetType: data[0],
bodySize: C_AMF_DecodeInt24(data[1:4]),
timestamp: C_AMF_DecodeInt24(data[4:7]) | uint32(data[7])<<24,
channel: chanSource,
info: s.streamID,
}
pkt.resize(pkt.bodySize, headerSizeAuto)
copy(pkt.body, data[minDataSize:minDataSize+pkt.bodySize])
err := pkt.write(s, false)
if err != nil { if err != nil {
return 0, err return 0, err
} }
return len(data), nil return len(data), nil
} }
// I/O functions
// read from an RTMP connection. Sends a bytes received message if the
// number of bytes received (nBytesIn) is greater than the number sent
// (nBytesInSent) by 10% of the bandwidth.
func (s *Session) read(buf []byte) (int, error) {
err := s.link.conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(s.link.timeout)))
if err != nil {
return 0, err
}
n, err := io.ReadFull(s.link.conn, buf)
if err != nil {
s.log(DebugLevel, pkg+"read failed", "error", err.Error())
return 0, err
}
s.nBytesIn += int32(n)
if s.nBytesIn > (s.nBytesInSent + s.clientBW/10) {
err := sendBytesReceived(s)
if err != nil {
return n, err // NB: we still read n bytes, even though send bytes failed
}
}
return n, nil
}
// write to an RTMP connection.
func (s *Session) write(buf []byte) (int, error) {
//ToDo: consider using a different timeout for writes than for reads
err := s.link.conn.SetWriteDeadline(time.Now().Add(time.Second * time.Duration(s.link.timeout)))
if err != nil {
return 0, err
}
n, err := s.link.conn.Write(buf)
if err != nil {
s.log(WarnLevel, pkg+"write failed", "error", err.Error())
return 0, err
}
return n, nil
}
// isConnected returns true if the RTMP connection is up. // isConnected returns true if the RTMP connection is up.
func (s *Session) isConnected() bool { func (s *Session) isConnected() bool {
return s.link.conn != nil return s.link.conn != nil
@ -167,5 +245,5 @@ func (s *Session) isConnected() bool {
// enableWrite enables the current session for writing. // enableWrite enables the current session for writing.
func (s *Session) enableWrite() { func (s *Session) enableWrite() {
s.link.protocol |= RTMP_FEATURE_WRITE s.link.protocol |= featureWrite
} }

103
stream/mts/psi/psi.go
View File

@ -114,109 +114,6 @@ type Desc struct {
Dd []byte // Descriptor data Dd []byte // Descriptor data
} }
// ReadPSI creates a PSI data structure from a given byte slice that represents a PSI
func ReadPSI(data []byte) *PSI {
psi := PSI{}
pos := 0
psi.Pf = data[pos]
if psi.Pf != 0 {
panic("No support for pointer filler bytes")
}
psi.Tid = data[pos]
pos++
psi.Ssi = data[pos]&0x80 != 0
psi.Pb = data[pos]&0x40 != 0
psi.Sl = uint16(data[pos]&0x03)<<8 | uint16(data[pos+1])
pos += 2
psi.Tss = readTSS(data[pos:], &psi)
return &psi
}
// ReadTSS creates a TSS data structure from a given byte slice that represents a TSS
func readTSS(data []byte, p *PSI) *TSS {
tss := TSS{}
pos := 0
tss.Tide = uint16(data[pos])<<8 | uint16(data[pos+1])
pos += 2
tss.V = (data[pos] & 0x3e) >> 1
tss.Cni = data[pos]&0x01 != 0
pos++
tss.Sn = data[pos]
pos++
tss.Lsn = data[pos]
pos++
switch p.Tid {
case patID:
tss.Sd = readPAT(data[pos:], &tss)
case pmtID:
tss.Sd = readPMT(data[pos:], &tss)
default:
panic("Can't yet deal with tables that are not PAT or PMT")
}
return &tss
}
// readPAT creates a pat struct based on a bytes slice representing a pat
func readPAT(data []byte, p *TSS) *PAT {
pat := PAT{}
pos := 0
pat.Pn = uint16(data[pos])<<8 | uint16(data[pos+1])
pos += 2
pat.Pmpid = uint16(data[pos]&0x1f)<<8 | uint16(data[pos+1])
return &pat
}
// readPMT creates a pmt struct based on a bytes slice that represents a pmt
func readPMT(data []byte, p *TSS) *PAT {
pmt := PMT{}
pos := 0
pmt.Pcrpid = uint16(data[pos]&0x1f)<<8 | uint16(data[pos+1])
pos += 2
pmt.Pil = uint16(data[pos]&0x03)<<8 | uint16(data[pos+1])
pos += 2
if pmt.Pil != 0 {
pmt.Pd = readDescs(data[pos:], int(pmt.Pil))
}
pos += int(pmt.Pil)
// TODO Read ES stuff
pmt.Essd = readEssd(data[pos:])
return nil
}
// readDescs reads provides a slice of Descs given a byte slice that represents Descs
// and the no of bytes that the descs accumilate
func readDescs(data []byte, descLen int) (o []Desc) {
pos := 0
o = make([]Desc, 1)
o[0].Dt = data[pos]
pos++
o[0].Dl = data[pos]
pos++
o[0].Dd = make([]byte, o[0].Dl)
for i := 0; i < int(o[0].Dl); i++ {
o[0].Dd[i] = data[pos]
pos++
}
if 2+len(o[0].Dd) != descLen {
panic("No support for reading more than one descriptor")
}
return
}
// readEESD creates an ESSD struct based on a bytes slice that represents ESSD
func readEssd(data []byte) *ESSD {
essd := ESSD{}
pos := 0
essd.St = data[pos]
pos++
essd.Epid = uint16(data[pos]&0x1f)<<8 | uint16(data[pos+1])
pos += 2
essd.Esil = uint16(data[pos]&0x03)<<8 | uint16(data[pos+1])
pos += 2
essd.Esd = readDescs(data[pos:], int(essd.Esil))
return &essd
}
// Bytes outputs a byte slice representation of the PSI // Bytes outputs a byte slice representation of the PSI
func (p *PSI) Bytes() []byte { func (p *PSI) Bytes() []byte {
out := make([]byte, 4) out := make([]byte, 4)