diff --git a/revid/senders.go b/revid/senders.go
index 9c9abca1..f09d34fa 100644
--- a/revid/senders.go
+++ b/revid/senders.go
@@ -264,7 +264,7 @@ func newRtmpSender(url string, timeout uint, retries int, log func(lvl int8, msg
 	var sess *rtmp.Session
 	var err error
 	for n := 0; n < retries; n++ {
-		sess = rtmp.NewSession(url, timeout)
+		sess = rtmp.NewSession(url, timeout, log)
 		err = sess.Open()
 		if err == nil {
 			break
@@ -310,7 +310,7 @@ func (s *rtmpSender) restart() error {
 		return err
 	}
 	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()
 		if err == nil {
 			break
diff --git a/rtmp/packet.go b/rtmp/packet.go
index 95efb658..6d33731c 100644
--- a/rtmp/packet.go
+++ b/rtmp/packet.go
@@ -35,8 +35,8 @@ LICENSE
 package rtmp
 
 import (
+	"io"
 	"encoding/binary"
-	"log"
 )
 
 const (
@@ -68,8 +68,12 @@ const (
 	RTMP_CHANNEL_SOURCE     = 0x04
 )
 
-// packetSize defines valid packet sizes.
-var packetSize = [...]int{12, 8, 4, 1}
+// headerSizes defines header sizes for header types 0, 1, 2 and 3 respectively:
+//   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.
 type packet struct {
@@ -101,7 +105,10 @@ func readPacket(s *Session, pkt *packet) error {
 
 	err := readN(s, header[:1])
 	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
 	}
 	pkt.headerType = (header[0] & 0xc0) >> 6
@@ -112,7 +119,7 @@ func readPacket(s *Session, pkt *packet) error {
 	case pkt.channel == 0:
 		err = readN(s, header[:1])
 		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
 		}
 		header = header[1:]
@@ -121,12 +128,11 @@ func readPacket(s *Session, pkt *packet) error {
 	case pkt.channel == 1:
 		err = readN(s, header[:2])
 		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
 		}
 		header = header[2:]
 		pkt.channel = int32(binary.BigEndian.Uint16(header[:2])) + 64
-
 	}
 
 	if pkt.channel >= s.channelsAllocatedIn {
@@ -134,49 +140,46 @@ func readPacket(s *Session, pkt *packet) error {
 		timestamp := append(s.channelTimestamp, make([]int32, 10)...)
 
 		var pkts []*packet
-		if s.vecChannelsIn == nil {
+		if s.channelsIn == nil {
 			pkts = make([]*packet, n)
 		} 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.vecChannelsIn = pkts
+		s.channelsIn = pkts
 
 		for i := int(s.channelsAllocatedIn); i < len(s.channelTimestamp); i++ {
 			s.channelTimestamp[i] = 0
 		}
 		for i := int(s.channelsAllocatedIn); i < int(n); i++ {
-			s.vecChannelsIn[i] = nil
+			s.channelsIn[i] = nil
 		}
 		s.channelsAllocatedIn = n
 	}
 
-	size := packetSize[pkt.headerType]
+	size := headerSizes[pkt.headerType]
 	switch {
 	case size == RTMP_LARGE_HEADER_SIZE:
 		pkt.hasAbsTimestamp = true
 	case size < RTMP_LARGE_HEADER_SIZE:
-		if s.vecChannelsIn[pkt.channel] != nil {
-			*pkt = *(s.vecChannelsIn[pkt.channel])
+		if s.channelsIn[pkt.channel] != nil {
+			*pkt = *(s.channelsIn[pkt.channel])
 		}
 	}
-
 	size--
 
 	if size > 0 {
 		err = readN(s, header[:size])
 		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
 		}
 	}
-
 	hSize := len(hbuf) - len(header) + size
 
 	if size >= 3 {
 		pkt.timestamp = C_AMF_DecodeInt24(header[:3])
-
 		if size >= 6 {
 			pkt.bodySize = C_AMF_DecodeInt24(header[3:6])
 			pkt.bytesRead = 0
@@ -195,7 +198,7 @@ func readPacket(s *Session, pkt *packet) error {
 	if extendedTimestamp {
 		err = readN(s, header[size:size+4])
 		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
 		}
 		// TODO: port this
@@ -222,20 +225,20 @@ func readPacket(s *Session, pkt *packet) error {
 
 	err = readN(s, pkt.body[pkt.bytesRead:][:chunkSize])
 	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
 	}
 
 	pkt.bytesRead += uint32(chunkSize)
 
 	// keep the packet as ref for other packets on this channel
-	if s.vecChannelsIn[pkt.channel] == nil {
-		s.vecChannelsIn[pkt.channel] = &packet{}
+	if s.channelsIn[pkt.channel] == nil {
+		s.channelsIn[pkt.channel] = &packet{}
 	}
-	*(s.vecChannelsIn[pkt.channel]) = *pkt
+	*(s.channelsIn[pkt.channel]) = *pkt
 
 	if extendedTimestamp {
-		s.vecChannelsIn[pkt.channel].timestamp = 0xffffff
+		s.channelsIn[pkt.channel].timestamp = 0xffffff
 	}
 
 	if pkt.bytesRead != pkt.bodySize {
@@ -248,13 +251,13 @@ func readPacket(s *Session, pkt *packet) error {
 	}
 	s.channelTimestamp[pkt.channel] = int32(pkt.timestamp)
 
-	s.vecChannelsIn[pkt.channel].body = nil
-	s.vecChannelsIn[pkt.channel].bytesRead = 0
-	s.vecChannelsIn[pkt.channel].hasAbsTimestamp = false
+	s.channelsIn[pkt.channel].body = nil
+	s.channelsIn[pkt.channel].bytesRead = 0
+	s.channelsIn[pkt.channel].hasAbsTimestamp = false
 	return nil
 }
 
-// resizePacket adjust the packet's storage to accommodate a body of the given size.
+// resizePacket adjusts the packet's storage to accommodate a body of the given size.
 func resizePacket(pkt *packet, size uint32, ht uint8) {
 	buf := make([]byte, RTMP_MAX_HEADER_SIZE+size)
 	pkt.headerType = ht
@@ -271,20 +274,20 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
 		n := int(pkt.channel + 10)
 
 		var pkts []*packet
-		if s.vecChannelsOut == nil {
+		if s.channelsOut == nil {
 			pkts = make([]*packet, n)
 		} 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++ {
-			s.vecChannelsOut[i] = nil
+			s.channelsOut[i] = nil
 		}
 
 		s.channelsAllocatedOut = int32(n)
 	}
-	prevPkt = s.vecChannelsOut[pkt.channel]
+	prevPkt = s.channelsOut[pkt.channel]
 
 	if prevPkt != nil && pkt.headerType != RTMP_PACKET_SIZE_LARGE {
 		// compress a bit by using the prev packet's attributes
@@ -300,8 +303,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
 	}
 
 	if pkt.headerType > 3 {
-		log.Printf("Sanity failed! trying to send header of type: 0x%02x.",
-			pkt.headerType)
+		s.log(WarnLevel, pkg+"unexpected header type", "type", pkt.headerType)
 		return errInvalidHeader
 	}
 
@@ -310,7 +312,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
 	if pkt.body != nil {
 		// Span from -packetsize for the type to the start of the body.
 		headBytes = pkt.header
-		origIdx = RTMP_MAX_HEADER_SIZE - packetSize[pkt.headerType]
+		origIdx = RTMP_MAX_HEADER_SIZE - headerSizes[pkt.headerType]
 	} else {
 		// Allocate a new header and allow 6 bytes of movement backward.
 		var hbuf [RTMP_MAX_HEADER_SIZE]byte
@@ -326,7 +328,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
 		cSize = 1
 	}
 
-	hSize := packetSize[pkt.headerType]
+	hSize := headerSizes[pkt.headerType]
 	if cSize != 0 {
 		origIdx -= cSize
 		hSize += cSize
@@ -339,7 +341,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
 	if ts >= 0xffffff {
 		origIdx -= 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
@@ -367,7 +369,7 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
 		}
 	}
 
-	if packetSize[pkt.headerType] > 1 {
+	if headerSizes[pkt.headerType] > 1 {
 		res := ts
 		if ts > 0xffffff {
 			res = 0xffffff
@@ -376,14 +378,14 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
 		headerIdx += 3 // 24bits
 	}
 
-	if packetSize[pkt.headerType] > 4 {
+	if headerSizes[pkt.headerType] > 4 {
 		C_AMF_EncodeInt24(headBytes[headerIdx:], int32(pkt.bodySize))
 		headerIdx += 3 // 24bits
 		headBytes[headerIdx] = pkt.packetType
 		headerIdx++
 	}
 
-	if packetSize[pkt.headerType] > 8 {
+	if headerSizes[pkt.headerType] > 8 {
 		n := int(encodeInt32LE(headBytes[headerIdx:headerIdx+4], pkt.info))
 		headerIdx += n
 	}
@@ -396,39 +398,37 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
 	size := int(pkt.bodySize)
 	chunkSize := int(s.outChunkSize)
 
-	if debugMode {
-		log.Printf("sendPacket: %v->%v, size=%v", s.link.conn.LocalAddr(), s.link.conn.RemoteAddr(), size)
-	}
+	s.log(DebugLevel, pkg+"sending packet", "size", size, "la", s.link.conn.LocalAddr(), "ra", s.link.conn.RemoteAddr())
 
-	// Send the previously deferred packet if combining it with the next packet would exceed the chunk size.
-	if s.defered != nil && len(s.defered)+size+hSize > chunkSize {
-		err := writeN(s, s.defered)
+	if s.deferred != nil && len(s.deferred)+size+hSize > chunkSize {
+		err := writeN(s, s.deferred)
 		if err != nil {
 			return err
 		}
-		s.defered = nil
+		s.deferred = nil
 	}
 
 	// 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.
 	for size+hSize != 0 {
-		if s.defered == nil && pkt.packetType == RTMP_PACKET_TYPE_AUDIO && size < chunkSize {
-			s.defered = headBytes[origIdx:][:size+hSize]
+		if s.deferred == nil && pkt.packetType == RTMP_PACKET_TYPE_AUDIO && size < chunkSize {
+			s.deferred = headBytes[origIdx:][:size+hSize]
+			s.log(DebugLevel, pkg+"deferred sending packet")
 			break
 		}
 		if chunkSize > size {
 			chunkSize = size
 		}
 		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.
-			bytes = append(s.defered, bytes...)
+			bytes = append(s.deferred, bytes...)
 		}
 		err := writeN(s, bytes)
 		if err != nil {
 			return err
 		}
-		s.defered = nil
+		s.deferred = nil
 
 		size -= chunkSize
 		origIdx += chunkSize + hSize
@@ -464,22 +464,19 @@ func sendPacket(s *Session, pkt *packet, queue bool) error {
 	if pkt.packetType == RTMP_PACKET_TYPE_INVOKE {
 		buf := pkt.body[1:]
 		meth := C_AMF_DecodeString(buf)
-
-		if debugMode {
-			log.Printf("invoking %v", meth)
-		}
 		// keep it in call queue till result arrives
 		if queue {
+			s.log(DebugLevel, pkg+"queuing method "+meth)
 			buf = buf[3+len(meth):]
 			txn := int32(C_AMF_DecodeNumber(buf[:8]))
 			s.methodCalls = append(s.methodCalls, method{name: meth, num: txn})
 		}
 	}
 
-	if s.vecChannelsOut[pkt.channel] == nil {
-		s.vecChannelsOut[pkt.channel] = &packet{}
+	if s.channelsOut[pkt.channel] == nil {
+		s.channelsOut[pkt.channel] = &packet{}
 	}
-	*(s.vecChannelsOut[pkt.channel]) = *pkt
+	*(s.channelsOut[pkt.channel]) = *pkt
 
 	return nil
 }
diff --git a/rtmp/rtmp.go b/rtmp/rtmp.go
index d73c2fc2..2c6300d5 100644
--- a/rtmp/rtmp.go
+++ b/rtmp/rtmp.go
@@ -38,9 +38,7 @@ import (
 	"bytes"
 	"encoding/binary"
 	"errors"
-	"fmt"
 	"io"
-	"log"
 	"math/rand"
 	"net"
 	"strconv"
@@ -48,9 +46,8 @@ import (
 )
 
 const (
+	pkg         = "rtmp:"
 	minDataSize = 11
-	debugMode   = false
-	length      = 512
 )
 
 const (
@@ -123,6 +120,7 @@ var rtmpProtocolStrings = [...]string{
 // RTMP errors.
 var (
 	errUnknownScheme = errors.New("rtmp: unknown scheme")
+	errConnected     = errors.New("rtmp: already connected")
 	errNotConnected  = errors.New("rtmp: not connected")
 	errHandshake     = errors.New("rtmp: handshake failed")
 	errConnSend      = errors.New("rtmp: connection send error")
@@ -144,8 +142,7 @@ func setupURL(s *Session, addr string) (err error) {
 
 	if s.link.tcUrl == "" {
 		if s.link.app != "" {
-			s.link.tcUrl = fmt.Sprintf("%v://%v:%v/%v",
-				rtmpProtocolStrings[s.link.protocol], s.link.host, s.link.port, s.link.app)
+			s.link.tcUrl = rtmpProtocolStrings[s.link.protocol] + "://" + s.link.host + ":" + strconv.Itoa(int(s.link.port)) + "/" + s.link.app
 			s.link.lFlags |= RTMP_LF_FTCU
 		} else {
 			s.link.tcUrl = addr
@@ -156,6 +153,7 @@ func setupURL(s *Session, addr string) (err error) {
 		switch {
 		case (s.link.protocol & RTMP_FEATURE_SSL) != 0:
 			s.link.port = 433
+			s.log(FatalLevel, pkg+"SSL not supported")
 		case (s.link.protocol & RTMP_FEATURE_HTTP) != 0:
 			s.link.port = 80
 		default:
@@ -173,22 +171,19 @@ func connect(s *Session) error {
 	}
 	s.link.conn, err = net.DialTCP("tcp4", nil, addr)
 	if err != nil {
+		s.log(WarnLevel, pkg+"dial failed", "error", err.Error())
 		return err
 	}
-	if debugMode {
-		log.Println("... connected, handshaking...")
-	}
+	s.log(DebugLevel, pkg+"connected")
 	err = handshake(s)
 	if err != nil {
-		log.Println("connect: handshake failed")
+		s.log(WarnLevel, pkg+"handshake failed", "error", err.Error())
 		return errHandshake
 	}
-	if debugMode {
-		log.Println("... handshaked...")
-	}
+	s.log(DebugLevel, pkg+"handshaked")
 	err = sendConnectPacket(s)
 	if err != nil {
-		log.Println("connect: sendConnect failed")
+		s.log(WarnLevel, pkg+"sendConnect failed", "error", err.Error())
 		return errConnSend
 	}
 	return nil
@@ -196,40 +191,38 @@ func connect(s *Session) error {
 
 // connectStream reads a packet and handles it
 func connectStream(s *Session) error {
-	var pkt packet
-
+	var err error
 	for !s.isPlaying && s.isConnected() {
-		err := readPacket(s, &pkt)
+		pkt := packet{}
+		err = readPacket(s, &pkt)
 		if err != nil {
 			break
 		}
-
 		if pkt.bodySize == 0 {
 			continue
 		}
-
 		if pkt.packetType == RTMP_PACKET_TYPE_AUDIO ||
 			pkt.packetType == RTMP_PACKET_TYPE_VIDEO ||
 			pkt.packetType == RTMP_PACKET_TYPE_INFO {
-			log.Println("connectStream: got packet before play()! Ignoring.")
-			pkt.body = nil
+			s.log(DebugLevel, pkg+"got packet before play; ignoring")
 			continue
 		}
 
-		handlePacket(s, &pkt)
-		pkt.body = nil
+		err = handlePacket(s, &pkt)
+		if err != nil {
+			break
+		}
 	}
 
 	if !s.isPlaying {
-		return errConnStream
+		return err
 	}
 	return nil
 }
 
 // handlePacket handles a packet that the client has received.
 // NB: cases have been commented out that are not currently used by AusOcean
-func handlePacket(s *Session, pkt *packet) int32 {
-	var hasMediaPacket int32
+func handlePacket(s *Session, pkt *packet) error {
 	switch pkt.packetType {
 
 	case RTMP_PACKET_TYPE_CHUNK_SIZE:
@@ -241,7 +234,7 @@ func handlePacket(s *Session, pkt *packet) int32 {
 		s.serverBW = int32(C_AMF_DecodeInt32(pkt.body[:4]))
 
 	case RTMP_PACKET_TYPE_CONTROL:
-		panic("Unsupported packet type RTMP_PACKET_TYPE_CONTROL")
+		s.log(FatalLevel, pkg+"unsupported packet type RTMP_PACKET_TYPE_CONTROL")
 
 	case RTMP_PACKET_TYPE_SERVER_BW:
 		s.serverBW = int32(C_AMF_DecodeInt32(pkt.body[:4]))
@@ -255,36 +248,32 @@ func handlePacket(s *Session, pkt *packet) int32 {
 		}
 
 	case RTMP_PACKET_TYPE_AUDIO:
-		panic("Unsupported packet type RTMP_PACKET_TYPE_AUDIO")
+		s.log(FatalLevel, pkg+"unsupported packet type RTMP_PACKET_TYPE_AUDIO")
 
 	case RTMP_PACKET_TYPE_VIDEO:
-		panic("Unsupported packet type RTMP_PACKET_TYPE_VIDEO")
+		s.log(FatalLevel, pkg+"unsupported packet type RTMP_PACKET_TYPE_VIDEO")
 
 	case RTMP_PACKET_TYPE_FLEX_MESSAGE:
-		panic("Unsupported packet type RTMP_PACKET_TYPE_FLEX_MESSAGE")
+		s.log(FatalLevel, pkg+"unsupported packet type RTMP_PACKET_TYPE_FLEX_MESSAGE")
 
 	case RTMP_PACKET_TYPE_INFO:
-		panic("Unsupported packet type RTMP_PACKET_TYPE_INFO")
+		s.log(FatalLevel, pkg+"unsupported packet type RTMP_PACKET_TYPE_INFO")
 
 	case RTMP_PACKET_TYPE_INVOKE:
-		if debugMode {
-			log.Println("RTMP_PACKET_TYPE_INVOKE:")
-		}
 		err := handleInvoke(s, pkt.body[:pkt.bodySize])
 		if err != nil {
 			// This will never happen with the methods we implement.
-			log.Println("HasMediaPacket")
-			hasMediaPacket = 2
+			s.log(WarnLevel, pkg+"unexpected error from handleInvoke", "error", err.Error())
+			return err
 		}
 
 	case RTMP_PACKET_TYPE_FLASH_VIDEO:
-		panic("Unsupported packet type RTMP_PACKET_TYPE_FLASH_VIDEO")
+		s.log(FatalLevel, pkg+"unsupported packet type RTMP_PACKET_TYPE_FLASH_VIDEO")
 
 	default:
-		// TODO use new logger here
-		// RTMP_Log(RTMP_LOGDEBUG, "%s, unknown packet type received: 0x%02x", __FUNCTION__,pkt.packetType);
+		s.log(WarnLevel, pkg+"unknown packet type", "type", pkt.packetType)
 	}
-	return hasMediaPacket
+	return nil
 }
 
 func readN(s *Session, buf []byte) error {
@@ -294,10 +283,7 @@ func readN(s *Session, buf []byte) error {
 	}
 	n, err := io.ReadFull(s.link.conn, buf)
 	if err != nil {
-		if debugMode {
-			log.Printf("readN error: %v\n", err)
-		}
-		s.close()
+		s.log(DebugLevel, pkg+"read failed", "error", err.Error())
 		return err
 	}
 	s.nBytesIn += int32(n)
@@ -318,10 +304,7 @@ func writeN(s *Session, buf []byte) error {
 	}
 	_, err = s.link.conn.Write(buf)
 	if err != nil {
-		if debugMode {
-			log.Printf("writeN, RTMP send error: %v\n", err)
-		}
-		s.close()
+		s.log(WarnLevel, pkg+"write failed", "error", err.Error())
 		return err
 	}
 	return nil
@@ -417,12 +400,10 @@ func sendConnectPacket(s *Session) error {
 		}
 	}
 
-	if copy(enc, []byte{0, 0, AMF_OBJECT_END}) != 3 {
-		return errCopying // TODO: is this even possible?
-	}
+	copy(enc, []byte{0, 0, AMF_OBJECT_END})
 	enc = enc[3:]
 
-	/* add auth string */
+	// add auth string
 	if s.link.auth != "" {
 		enc = C_AMF_EncodeBoolean(enc, s.link.lFlags&RTMP_LF_AUTH != 0)
 		if enc == nil {
@@ -443,7 +424,7 @@ func sendConnectPacket(s *Session) error {
 
 	pkt.bodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
 
-	return sendPacket(s, &pkt, true)
+	return sendPacket(s, &pkt, true) // response expected
 }
 
 func sendCreateStream(s *Session) error {
@@ -471,7 +452,7 @@ func sendCreateStream(s *Session) error {
 
 	pkt.bodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
 
-	return sendPacket(s, &pkt, true)
+	return sendPacket(s, &pkt, true) // response expected
 }
 
 func sendReleaseStream(s *Session) error {
@@ -602,7 +583,7 @@ func sendPublish(s *Session) error {
 
 	pkt.bodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
 
-	return sendPacket(s, &pkt, true)
+	return sendPacket(s, &pkt, true) // response expected
 }
 
 func sendDeleteStream(s *Session, dStreamId float64) error {
@@ -706,9 +687,8 @@ func handleInvoke(s *Session, body []byte) error {
 	}
 
 	meth := C_AMFProp_GetString(C_AMF_GetProp(&obj, "", 0))
+	s.log(DebugLevel, pkg+"invoking method "+meth)
 	txn := C_AMFProp_GetNumber(C_AMF_GetProp(&obj, "", 1))
-	// TODO use new logger here
-	// RTMP_Log(RTMP_LOGDEBUG, "%s, server invoking <%s>", __FUNCTION__, method.av_val);
 
 	switch meth {
 	case av__result:
@@ -721,30 +701,27 @@ func handleInvoke(s *Session, body []byte) error {
 			}
 		}
 		if methodInvoked == "" {
-			// TODO use new logger here
-			//RTMP_Log(RTMP_LOGDEBUG, "%s, received result id %f without matching request",
-			//__FUNCTION__, txn);
+			s.log(WarnLevel, pkg+"received result without matching request", "id", txn)
 			goto leave
 		}
-		// TODO use new logger here
-		//RTMP_Log(RTMP_LOGDEBUG, "%s, received result for method call <%s>", __FUNCTION__,
-		//methodInvoked.av_val);
+		s.log(DebugLevel, pkg+"received result for "+methodInvoked)
+
 		switch methodInvoked {
 		case av_connect:
 			if s.link.token != "" {
-				panic("No support for link token")
+				s.log(FatalLevel, pkg+"no support for link token")
 
 			}
 			if (s.link.protocol & RTMP_FEATURE_WRITE) != 0 {
 				sendReleaseStream(s)
 				sendFCPublish(s)
 			} else {
-				panic("Link protocol has no RTMP_FEATURE_WRITE")
+				s.log(FatalLevel, pkg+"link protocol has no RTMP_FEATURE_WRITE")
 			}
 
 			sendCreateStream(s)
 			if (s.link.protocol & RTMP_FEATURE_WRITE) == 0 {
-				panic("Link protocol has no RTMP_FEATURE_WRITE")
+				s.log(FatalLevel, pkg+"link protocol has no RTMP_FEATURE_WRITE")
 			}
 
 		case av_createStream:
@@ -753,54 +730,50 @@ func handleInvoke(s *Session, body []byte) error {
 			if s.link.protocol&RTMP_FEATURE_WRITE != 0 {
 				sendPublish(s)
 			} else {
-				panic("Link protocol has no RTMP_FEATURE_WRITE")
+				s.log(FatalLevel, pkg+"link protocol has no RTMP_FEATURE_WRITE")
 			}
 
 		case av_play, av_publish:
-			panic("Unsupported method av_play/av_publish")
+			s.log(FatalLevel, pkg+"unsupported method av_play/av_publish")
 		}
-		//C.free(unsafe.Pointer(methodInvoked.av_val))
 
 	case av_onBWDone:
-		if s.bwCheckCounter == 0 {
+		if s.checkCounter == 0 { // ToDo: why is this always zero?
 			sendCheckBW(s)
 		}
 
 	case av_onFCUnsubscribe, av_onFCSubscribe:
-		panic("Unsupported method av_onFCUnsubscribe/av_onFCSubscribe")
+		s.log(FatalLevel, pkg+"unsupported method av_onFCUnsubscribe/av_onFCSubscribe")
 
 	case av_ping:
-		panic("Unsupported method av_ping")
+		s.log(FatalLevel, pkg+"unsupported method av_ping")
 
 	case av__onbwcheck:
-		panic("Unsupported method av_onbwcheck")
+		s.log(FatalLevel, pkg+"unsupported method av_onbwcheck")
 
 	case av__onbwdone:
-		panic("Unsupported method av_onbwdone")
+		s.log(FatalLevel, pkg+"unsupported method av_onbwdone")
 
 	case av_close:
-		panic("Unsupported method av_close")
+		s.log(FatalLevel, pkg+"unsupported method av_close")
 
 	case av_onStatus:
 		var obj2 C_AMFObject
 		C_AMFProp_GetObject(C_AMF_GetProp(&obj, "", 3), &obj2)
 		code := C_AMFProp_GetString(C_AMF_GetProp(&obj2, av_code, -1))
-
-		level := C_AMFProp_GetString(C_AMF_GetProp(&obj2, av_level, -1)) // Not used.
-		_ = level
-
-		// TODO use new logger
-		// RTMP_Log(RTMP_LOGDEBUG, "%s, onStatus: %s", __FUNCTION__, code.av_val);
+		level := C_AMFProp_GetString(C_AMF_GetProp(&obj2, av_level, -1))
+		s.log(DebugLevel, pkg+"onStatus", "code", code, "level", level)
 
 		switch code {
 		case av_NetStream_Failed, av_NetStream_Play_Failed,
 			av_NetStream_Play_StreamNotFound, av_NetConnection_Connect_InvalidApp:
-			panic("Unsupported method av_NetStream/av_NetStream_Play_Failed/av_netSTream_Play_StreamNotFound/av_netConnection_Connect_invalidApp")
+			s.log(FatalLevel, pkg+"unsupported method av_NetStream/av_NetStream_Play_Failed/av_netSTream_Play_StreamNotFound/av_netConnection_Connect_invalidApp")
 
 		case av_NetStream_Play_Start, av_NetStream_Play_PublishNotify:
-			panic("Unsupported method av_NetStream_Play_Start/av_NetStream_Play_PublishNotify")
+			s.log(FatalLevel, pkg+"unsupported method av_NetStream_Play_Start/av_NetStream_Play_PublishNotify")
 
 		case av_NetStream_Publish_Start:
+			s.log(DebugLevel, pkg+"playing")
 			s.isPlaying = true
 			for i, m := range s.methodCalls {
 				if m.name == av_publish {
@@ -808,22 +781,23 @@ func handleInvoke(s *Session, body []byte) error {
 					break
 				}
 			}
+			// ToDo: handle case when av_publish method not found
 
 		case av_NetStream_Play_Complete, av_NetStream_Play_Stop, av_NetStream_Play_UnpublishNotify:
-			panic("Unsupported method av_NetStream_Play_Complete/av_NetStream_Play_Stop/av_NetStream_Play_UnpublishNotify")
+			s.log(FatalLevel, pkg+"unsupported method av_NetStream_Play_Complete/av_NetStream_Play_Stop/av_NetStream_Play_UnpublishNotify")
 
 		case av_NetStream_Seek_Notify:
-			panic("Unsupported method av_netStream_Seek_Notify")
+			s.log(FatalLevel, pkg+"unsupported method av_netStream_Seek_Notify")
 
 		case av_NetStream_Pause_Notify:
-			panic("Unsupported method av_NetStream_Pause_Notify")
+			s.log(FatalLevel, pkg+"unsupported method av_NetStream_Pause_Notify")
 		}
 
 	case av_playlist_ready:
-		panic("Unsupported method av_playlist_ready")
+		s.log(FatalLevel, pkg+"unsupported method av_playlist_ready")
 
 	default:
-		panic(fmt.Sprintf("unknown method: %q", meth))
+		s.log(FatalLevel, pkg+"unknown method "+meth)
 	}
 leave:
 	C_AMF_Reset(&obj)
@@ -847,19 +821,17 @@ func handshake(s *Session) error {
 	if err != nil {
 		return err
 	}
+	s.log(DebugLevel, pkg+"handshake sent")
 
 	var typ [1]byte
 	err = readN(s, typ[:])
 	if err != nil {
 		return err
 	}
+	s.log(DebugLevel, pkg+"handshake received")
 
-	if debugMode {
-		log.Printf("handshake: Type answer: %v\n", typ[0])
-	}
 	if typ[0] != clientbuf[0] {
-		log.Printf("handshake: type mismatch: client sent %v, server sent: %v\n",
-			clientbuf[0], typ)
+		s.log(WarnLevel, pkg+"handshake type mismatch", "sent", clientbuf[0], "received", typ)
 	}
 	err = readN(s, serversig[:])
 	if err != nil {
@@ -868,10 +840,7 @@ func handshake(s *Session) error {
 
 	// decode server response
 	suptime := binary.BigEndian.Uint32(serversig[:4])
-	_ = suptime
-	// RTMP_Log(RTMP_LOGDEBUG, "%s: Server Uptime : %d", __FUNCTION__, suptime)
-	// RTMP_Log(RTMP_LOGDEBUG, "%s: FMS Version   : %d.%d.%d.%d", __FUNCTION__,
-	// 	serversig[4], serversig[5], serversig[6], serversig[7])
+	s.log(DebugLevel, pkg+"server uptime", "uptime", suptime)
 
 	// 2nd part of handshake
 	err = writeN(s, serversig[:])
@@ -885,8 +854,7 @@ func handshake(s *Session) error {
 	}
 
 	if !bytes.Equal(serversig[:RTMP_SIG_SIZE], clientbuf[1:RTMP_SIG_SIZE+1]) {
-		log.Printf("Client signature does not match: %q != %q",
-			serversig[:RTMP_SIG_SIZE], clientbuf[1:RTMP_SIG_SIZE+1])
+		s.log(WarnLevel, pkg+"signature mismatch", "serversig", serversig[:RTMP_SIG_SIZE], "clientsig", clientbuf[1:RTMP_SIG_SIZE+1])
 	}
 	return nil
 }
diff --git a/rtmp/rtmp_test.go b/rtmp/rtmp_test.go
new file mode 100644
index 00000000..296809c6
--- /dev/null
+++ b/rtmp/rtmp_test.go
@@ -0,0 +1,217 @@
+/*
+NAME
+  rtmp_test.go
+
+DESCRIPTION
+  RTMP tests
+
+AUTHORS
+  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 RTMP key required for YouTube streaming (RTMP_TEST_KEY env var)
+var testKey string
+
+// testFile is the test video file  (RTMP_TEST_FILE env var)
+var testFile string
+
+// testLog is a bare bones logger that logs to stdout.
+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")
+	}
+	if testVerbosity == 2 && len(params) >= 2 {
+		// extract the params we know about, otherwise just print the message
+		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 == 5 {
+		// 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, s.url)
+	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
+	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)
+		}
+	}
+
+	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)
+	}
+}
diff --git a/rtmp/session.go b/rtmp/session.go
index 6d1fad40..a402a320 100644
--- a/rtmp/session.go
+++ b/rtmp/session.go
@@ -33,17 +33,12 @@ LICENSE
 */
 package rtmp
 
-import (
-	"errors"
-)
-
 // Session holds the state for an RTMP session.
 type Session struct {
 	url                  string
-	timeout              uint
 	inChunkSize          int32
 	outChunkSize         int32
-	bwCheckCounter       int32
+	checkCounter         int32
 	nBytesIn             int32
 	nBytesInSent         int32
 	streamID             int32
@@ -56,28 +51,52 @@ type Session struct {
 	methodCalls          []method
 	channelsAllocatedIn  int32
 	channelsAllocatedOut int32
-	vecChannelsIn        []*packet
-	vecChannelsOut       []*packet
+	channelsIn           []*packet
+	channelsOut          []*packet
 	channelTimestamp     []int32
 	audioCodecs          float64
 	videoCodecs          float64
 	encoding             float64
-	defered              []byte
+	deferred             []byte
 	link                 link
+	log                  Log
 }
 
+// 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.
-func NewSession(url string, connectTimeout uint) *Session {
+func NewSession(url string, timeout uint, log Log) *Session {
 	return &Session{
-		url:     url,
-		timeout: connectTimeout,
+		url:          url,
+		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.
 func (s *Session) Open() error {
 	if s.isConnected() {
-		return errors.New("rtmp: attempt to start already running session")
+		return errConnected
 	}
 	err := s.start()
 	if err != nil {
@@ -88,7 +107,7 @@ func (s *Session) Open() error {
 
 // start does the heavylifting for Open().
 func (s *Session) start() error {
-	s.init()
+	s.log(DebugLevel, pkg+"Session.start")
 	err := setupURL(s, s.url)
 	if err != nil {
 		s.close()
@@ -110,20 +129,6 @@ func (s *Session) start() error {
 	return nil
 }
 
-// init initializes various RTMP defauls.
-// ToDo: define consts for the magic numbers.
-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 {
 	if !s.isConnected() {
@@ -136,6 +141,7 @@ func (s *Session) Close() error {
 // 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() {
+	s.log(DebugLevel, pkg+"Session.close")
 	if s.isConnected() {
 		if s.streamID > 0 {
 			if s.link.protocol&RTMP_FEATURE_WRITE != 0 {