Make srtconn.ReadPacket and srtconn.WritePacket public

connection.go

@@ -25,11 +25,20 @@ type Conn interface {
 	// time limit; see SetDeadline and SetReadDeadline.
 	Read(p []byte) (int, error)
 
+	// ReadPacket reads a packet from the queue of received packets. It blocks
+	// if the queue is empty. Only data packets are returned. Using ReadPacket
+	// and Read at the same time may lead to data loss.
+	ReadPacket() (packet.Packet, error)
+
 	// Write writes data to the connection.
 	// Write can be made to time out and return an error after a fixed
 	// time limit; see SetDeadline and SetWriteDeadline.
 	Write(p []byte) (int, error)
 
+	// WritePacket writes a packet to the write queue. Packets on the write queue
+	// will be sent to the peer of the connection. Only data packets will be sent.
+	WritePacket(p packet.Packet) error
+
 	// Close closes the connection.
 	// Any blocked Read or Write operations will be unblocked and return errors.
 	Close() error
@@ -367,9 +376,7 @@ func (c *srtConn) ticker(ctx context.Context) {
 	}
 }
 
-// readPacket reads a packet from the queue of received packets. It blocks
-// if the queue is empty. Only data packets are returned.
-func (c *srtConn) readPacket() (packet.Packet, error) {
+func (c *srtConn) ReadPacket() (packet.Packet, error) {
 	var p packet.Packet
 	select {
 	case <-c.ctx.Done():
@@ -400,7 +407,7 @@ func (c *srtConn) Read(b []byte) (int, error) {
 
 	c.readBuffer.Reset()
 
-	p, err := c.readPacket()
+	p, err := c.ReadPacket()
 	if err != nil {
 		return 0, err
 	}
@@ -413,9 +420,9 @@ func (c *srtConn) Read(b []byte) (int, error) {
 	return c.readBuffer.Read(b)
 }
 
-// writePacket writes a packet to the write queue. Packets on the write queue
+// WritePacket writes a packet to the write queue. Packets on the write queue
 // will be sent to the peer of the connection. Only data packets will be sent.
-func (c *srtConn) writePacket(p packet.Packet) error {
+func (c *srtConn) WritePacket(p packet.Packet) error {
 	if p.Header().IsControlPacket {
 		// Ignore control packets
 		return nil

dial.go

@@ -701,7 +701,7 @@ func (dl *dialer) Read(p []byte) (n int, err error) {
 	return dl.conn.Read(p)
 }
 
-func (dl *dialer) readPacket() (packet.Packet, error) {
+func (dl *dialer) ReadPacket() (packet.Packet, error) {
 	if err := dl.checkConnection(); err != nil {
 		return nil, err
 	}
@@ -713,7 +713,7 @@ func (dl *dialer) readPacket() (packet.Packet, error) {
 		return nil, fmt.Errorf("no connection")
 	}
 
-	return dl.conn.readPacket()
+	return dl.conn.ReadPacket()
 }
 
 func (dl *dialer) Write(p []byte) (n int, err error) {
@@ -731,7 +731,7 @@ func (dl *dialer) Write(p []byte) (n int, err error) {
 	return dl.conn.Write(p)
 }
 
-func (dl *dialer) writePacket(p packet.Packet) error {
+func (dl *dialer) WritePacket(p packet.Packet) error {
 	if err := dl.checkConnection(); err != nil {
 		return err
 	}
@@ -743,7 +743,7 @@ func (dl *dialer) writePacket(p packet.Packet) error {
 		return fmt.Errorf("no connection")
 	}
 
-	return dl.conn.writePacket(p)
+	return dl.conn.WritePacket(p)
 }
 
 func (dl *dialer) SetDeadline(t time.Time) error      { return dl.conn.SetDeadline(t) }

pubsub.go

@@ -23,11 +23,6 @@ type PubSub interface {
 	Subscribe(c Conn) error
 }
 
-type packetReadWriter interface {
-	readPacket() (packet.Packet, error)
-	writePacket(p packet.Packet) error
-}
-
 // pubSub is an implementation of the PubSub interface
 type pubSub struct {
 	incoming      chan packet.Packet
@@ -107,12 +102,6 @@ func (pb *pubSub) Publish(c Conn) error {
 
 	var p packet.Packet
 	var err error
-	conn, ok := c.(packetReadWriter)
-	if !ok {
-		err := fmt.Errorf("the provided connection is not a SRT connection")
-		pb.logger.Print("pubsub:error", 0, 1, func() string { return err.Error() })
-		return err
-	}
 
 	socketId := c.SocketId()
 
@@ -121,7 +110,7 @@ func (pb *pubSub) Publish(c Conn) error {
 	pb.publish = true
 
 	for {
-		p, err = conn.readPacket()
+		p, err = c.ReadPacket()
 		if err != nil {
 			pb.logger.Print("pubsub:error", socketId, 1, func() string { return err.Error() })
 			break
@@ -142,12 +131,6 @@ func (pb *pubSub) Publish(c Conn) error {
 func (pb *pubSub) Subscribe(c Conn) error {
 	l := make(chan packet.Packet, 1024)
 	socketId := c.SocketId()
-	conn, ok := c.(packetReadWriter)
-	if !ok {
-		err := fmt.Errorf("the provided connection is not a SRT connection")
-		pb.logger.Print("pubsub:error", 0, 1, func() string { return err.Error() })
-		return err
-	}
 
 	pb.logger.Print("pubsub:subscribe", socketId, 1, func() string { return "new subscriber" })
 
@@ -166,7 +149,7 @@ func (pb *pubSub) Subscribe(c Conn) error {
 		case <-pb.ctx.Done():
 			return io.EOF
 		case p := <-l:
-			err := conn.writePacket(p)
+			err := c.WritePacket(p)
 			p.Decommission()
 			if err != nil {
 				pb.logger.Print("pubsub:error", socketId, 1, func() string { return err.Error() })