Improved debug logging.

client.go

@@ -133,7 +133,6 @@ func (c *Client) NewSession(opts ...SessionOption) (*Session, error) {
 		OutgoingWindow: s.outgoingWindow,
 		HandleMax:      s.handleMax,
 	}
-	debug(1, "TX: %s", begin)
 	s.txFrame(begin, nil)
 
 	// wait for response
@@ -143,8 +142,6 @@ func (c *Client) NewSession(opts ...SessionOption) (*Session, error) {
 		return nil, c.conn.getErr()
 	case fr = <-s.rx:
 	}
-	debug(1, "RX: %s", fr.body)
-
 	begin, ok := fr.body.(*performBegin)
 	if !ok {
 		_ = s.Close(context.Background()) // deallocate session on error
@@ -564,8 +561,6 @@ func (s *Session) mux(remoteBegin *performBegin) {
 
 		// incoming frame for link
 		case fr := <-s.rx:
-			debug(1, "RX(Session): %s", fr.body)
-
 			switch body := fr.body.(type) {
 			// Disposition frames can reference transfers from more than one
 			// link. Send this frame to all of them.
@@ -660,7 +655,6 @@ func (s *Session) mux(remoteBegin *performBegin) {
 						NextOutgoingID: nextOutgoingID,
 						OutgoingWindow: s.outgoingWindow,
 					}
-					debug(1, "TX: %s", resp)
 					s.txFrame(resp, nil)
 				}
 
@@ -712,7 +706,6 @@ func (s *Session) mux(remoteBegin *performBegin) {
 						NextOutgoingID: nextOutgoingID,
 						OutgoingWindow: s.outgoingWindow,
 					}
-					debug(1, "TX(Session): %s", flow)
 					s.txFrame(flow, nil)
 					remoteOutgoingWindow = s.incomingWindow
 				}
@@ -762,8 +755,6 @@ func (s *Session) mux(remoteBegin *performBegin) {
 				settlementByDeliveryID[deliveryID] = fr.done
 				fr.done = nil
 			}
-
-			debug(2, "TX(Session): %s", fr)
 			s.txFrame(fr, fr.done)
 
 			// "Upon sending a transfer, the sending endpoint will increment
@@ -780,13 +771,11 @@ func (s *Session) mux(remoteBegin *performBegin) {
 				fr.IncomingWindow = s.incomingWindow
 				fr.NextOutgoingID = nextOutgoingID
 				fr.OutgoingWindow = s.outgoingWindow
-				debug(1, "TX(Session): %s", fr)
 				s.txFrame(fr, nil)
 				remoteOutgoingWindow = s.incomingWindow
 			case *performTransfer:
 				panic("transfer frames must use txTransfer")
 			default:
-				debug(1, "TX(Session): %s", fr)
 				s.txFrame(fr, nil)
 			}
 		}
@@ -936,7 +925,6 @@ func attachLink(s *Session, r *Receiver, opts []LinkOption) (*link, error) {
 	}
 
 	// send Attach frame
-	debug(1, "TX: %s", attach)
 	s.txFrame(attach, nil)
 
 	// wait for response
@@ -946,7 +934,6 @@ func attachLink(s *Session, r *Receiver, opts []LinkOption) (*link, error) {
 		return nil, s.err
 	case fr = <-l.rx:
 	}
-	debug(3, "RX: %s", fr)
 	resp, ok := fr.(*performAttach)
 	if !ok {
 		return nil, errorErrorf("unexpected attach response: %#v", fr)
@@ -979,7 +966,6 @@ func attachLink(s *Session, r *Receiver, opts []LinkOption) (*link, error) {
 			Handle: l.handle,
 			Closed: true,
 		}
-		debug(1, "TX: %s", fr)
 		s.txFrame(fr, nil)
 
 		if detach.Error == nil {
@@ -1105,8 +1091,6 @@ Loop:
 
 		// send data
 		case tr := <-outgoingTransfers:
-			debug(3, "TX(link): %s", tr)
-
 			// Ensure the session mux is not blocked
 			for {
 				select {
@@ -1156,7 +1140,6 @@ func (l *link) muxFlow() error {
 		DeliveryCount: &deliveryCount,
 		LinkCredit:    &linkCredit, // max number of messages
 	}
-	debug(3, "TX: %s", fr)
 
 	// Update credit. This must happen before entering loop below
 	// because incoming messages handled while waiting to transmit
@@ -1323,7 +1306,6 @@ func (l *link) muxHandleFrame(fr frameBody) error {
 	switch fr := fr.(type) {
 	// message frame
 	case *performTransfer:
-		debug(3, "RX: %s", fr)
 		if isSender {
 			// Senders should never receive transfer frames, but handle it just in case.
 			l.closeWithError(&Error{
@@ -1337,7 +1319,6 @@ func (l *link) muxHandleFrame(fr frameBody) error {
 
 	// flow control frame
 	case *performFlow:
-		debug(3, "RX: %s", fr)
 		if isSender {
 			linkCredit := *fr.LinkCredit - l.deliveryCount
 			if fr.DeliveryCount != nil {
@@ -1365,12 +1346,10 @@ func (l *link) muxHandleFrame(fr frameBody) error {
 			DeliveryCount: &deliveryCount,
 			LinkCredit:    &linkCredit, // max number of messages
 		}
-		debug(1, "TX: %s", resp)
 		l.session.txFrame(resp, nil)
 
 	// remote side is closing links
 	case *performDetach:
-		debug(1, "RX: %s", fr)
 		// don't currently support link detach and reattach
 		if !fr.Closed {
 			return errorErrorf("non-closing detach not supported: %+v", fr)
@@ -1382,8 +1361,6 @@ func (l *link) muxHandleFrame(fr frameBody) error {
 		return errorWrapf(&DetachError{fr.Error}, "received detach frame")
 
 	case *performDisposition:
-		debug(3, "RX: %s", fr)
-
 		// Unblock receivers waiting for message disposition
 		if l.receiver != nil {
 			l.receiver.inFlight.remove(fr.First, fr.Last, nil)
@@ -1406,11 +1383,9 @@ func (l *link) muxHandleFrame(fr frameBody) error {
 			Last:    fr.Last,
 			Settled: true,
 		}
-		debug(1, "TX: %s", resp)
 		l.session.txFrame(resp, nil)
 
 	default:
-		debug(1, "RX: %s", fr)
 		fmt.Printf("Unexpected frame: %s\n", fr)
 	}
 
@@ -2006,8 +1981,6 @@ func (r *Receiver) sendDisposition(first uint32, last *uint32, state interface{}
 		Settled: r.link.receiverSettleMode == nil || *r.link.receiverSettleMode == ModeFirst,
 		State:   state,
 	}
-
-	debug(1, "TX: %s", fr)
 	return r.link.session.txFrame(fr, nil)
 }
 

conn.go

@@ -353,6 +353,7 @@ func (c *conn) mux() {
 
 		// new frame from connReader
 		case fr := <-c.rxFrame:
+			debugFrame(c, "RX", &fr)
 			var (
 				session *Session
 				ok      bool
@@ -619,7 +620,6 @@ func (c *conn) connWriter() {
 		case <-c.done:
 			// send close
 			cls := &performClose{}
-			debug(1, "TX: %s", cls)
 			_ = c.writeFrame(frame{
 				type_: frameTypeAMQP,
 				body:  cls,
@@ -632,6 +632,7 @@ func (c *conn) connWriter() {
 // writeFrame writes a frame to the network, may only be used
 // by connWriter after initial negotiation.
 func (c *conn) writeFrame(fr frame) error {
+	debugFrame(c, "TX", &fr)
 	if c.connectTimeout != 0 {
 		_ = c.net.SetWriteDeadline(time.Now().Add(c.connectTimeout))
 	}
@@ -807,7 +808,6 @@ func (c *conn) openAMQP() stateFunc {
 		IdleTimeout:  c.idleTimeout,
 		Properties:   c.properties,
 	}
-	debug(1, "TX: %s", open)
 	c.err = c.writeFrame(frame{
 		type_:   frameTypeAMQP,
 		body:    open,
@@ -828,7 +828,6 @@ func (c *conn) openAMQP() stateFunc {
 		c.err = errorErrorf("unexpected frame type %T", fr.body)
 		return nil
 	}
-	debug(1, "RX: %s", o)
 
 	// update peer settings
 	if o.MaxFrameSize > 0 {
@@ -860,7 +859,6 @@ func (c *conn) negotiateSASL() stateFunc {
 		c.err = errorErrorf("unexpected frame type %T", fr.body)
 		return nil
 	}
-	debug(1, "RX: %s", sm)
 
 	// return first match in c.saslHandlers based on order received
 	for _, mech := range sm.Mechanisms {
@@ -891,7 +889,6 @@ func (c *conn) saslOutcome() stateFunc {
 		c.err = errorErrorf("unexpected frame type %T", fr.body)
 		return nil
 	}
-	debug(1, "RX: %s", so)
 
 	// check if auth succeeded
 	if so.Code != codeSASLOK {
@@ -916,6 +913,7 @@ func (c *conn) readFrame() (frame, error) {
 	var fr frame
 	select {
 	case fr = <-c.rxFrame:
+		debugFrame(c, "RX", &fr)
 		return fr, nil
 	case err := <-c.connErr:
 		return fr, err

log.go

@@ -4,4 +4,5 @@ package amqp
 
 // dummy functions used when debugging is not enabled
 
-func debug(_ int, _ string, _ ...interface{}) {}
+func debug(_ int, _ string, _ ...interface{})      {}
+func debugFrame(c *conn, prefix string, fr *frame) {}

log_debug.go

@@ -25,3 +25,20 @@ func debug(level int, format string, v ...interface{}) {
 		logger.Printf(format, v...)
 	}
 }
+
+func debugFrame(c *conn, prefix string, fr *frame) {
+	if debugLevel == 0 { // Fast exit for no logging
+		return
+	}
+	// Set level by frame type
+	level := 1 // Normal frames at 1
+	switch fr.body.(type) {
+	case *performTransfer: // High-volume messages
+		level = 2
+	case *performFlow, *performDisposition: // Noisy flow and acknowledgment
+		level = 3
+	}
+	if level <= debugLevel {
+		logger.Printf("%p %s[%d]: %s", c, prefix, fr.channel, fr.body)
+	}
+}

sasl.go

@@ -47,7 +47,6 @@ func ConnSASLPlain(username, password string) ConnOption {
 				InitialResponse: []byte("\x00" + username + "\x00" + password),
 				Hostname:        "",
 			}
-			debug(1, "TX: %s", init)
 			c.err = c.writeFrame(frame{
 				type_: frameTypeSASL,
 				body:  init,
@@ -77,7 +76,6 @@ func ConnSASLAnonymous() ConnOption {
 				Mechanism:       saslMechanismANONYMOUS,
 				InitialResponse: []byte("anonymous"),
 			}
-			debug(1, "TX: %s", init)
 			c.err = c.writeFrame(frame{
 				type_: frameTypeSASL,
 				body:  init,