func sub(i int, wg *sync.WaitGroup) {
	topic := fmt.Sprintf("loadtest/%v", i)

	var cc *mqtt.ClientConn
	if cc = connect(); cc == nil {
		return
	}

	ack := cc.Subscribe([]proto.TopicQos{
		{topic, proto.QosAtLeastOnce},
	})
	if *dump {
		fmt.Printf("suback: %#v\n", ack)
	}

	go func() {
		ok := false
		count := 0
		for range cc.Incoming {
			count++
			if count == *messages {
				cc.Disconnect()
				ok = true
			}
		}

		if !ok {
			bad <- i
		}

		wg.Done()
	}()
}

func transmitter(topics []ifc.Topic, ss ss, cc *mqtt.ClientConn, mu *sync.Mutex) error {
	mu.Lock()
	suback := cc.Subscribe(tq(topics))
	mu.Unlock()

	if len(suback.TopicsQos) != len(topics) {
		return txerr(fmt.Errorf("suback has topic list length %v", len(suback.TopicsQos)))
	}

	// Since cc.Incoming is a channel, concurrent access to it is ok.
	for m1 := range cc.Incoming {
		if bp, ok := m1.Payload.(proto.BytesPayload); !ok {
			return txerr(fmt.Errorf("payload type %T not handled", m1.Payload))
		} else {
			m2 := ifc.Message{
				Topic:   m1.TopicName,
				Payload: []byte(bp),
			}
			err := ss.Send(m2)
			if err != nil {
				return txerr(err)
			}
		}
	}
	return nil
}

func receiver(rs rs, cc *mqtt.ClientConn, mu *sync.Mutex) error {
	for {
		if ok := rs.Advance(); !ok {
			return rxerr(rs.Err())
		}
		m1 := rs.Value()

		mu.Lock()
		cc.Publish(&proto.Publish{
			Header:    proto.Header{Retain: false},
			TopicName: string(m1.Topic),
			Payload:   proto.BytesPayload(m1.Payload),
		})
		mu.Unlock()
	}
}

func pub(i int) {
	topic := fmt.Sprintf("loadtest/%v", i)

	var cc *mqtt.ClientConn
	if cc = connect(); cc == nil {
		return
	}

	for i := 0; i < *messages; i++ {
		cc.Publish(&proto.Publish{
			Header:    proto.Header{QosLevel: proto.QosAtMostOnce},
			TopicName: topic,
			Payload:   proto.BytesPayload([]byte("loadtest payload")),
		})
	}

	cc.Disconnect()
}

func decode(cc *mqtt.ClientConn, src *net.UDPAddr, pkt []byte) {
	// Parse packet
	code := pkt[0]
	groupId := pkt[1]
	nodeId := pkt[2] & 0x1f
	ack := 0 // really need to decode pkt[2]
	data := pkt[3:]

	// Record the groupId -> addr mapping
	newGroup := saveGroupToAddr(groupId, src)

	// If this is a new group subscribe to the topic
	if newGroup {
		sub := []proto.TopicQos{
			{Topic: fmt.Sprintf("/rf/%d/+/tx", groupId), Qos: proto.QosAtMostOnce},
			{Topic: fmt.Sprintf("/rf/%d/+/tb", groupId), Qos: proto.QosAtMostOnce},
		}
		cc.Subscribe(sub)
	}

	// Create the topic
	if code > RF_Debug && code != RF_BootReply {
		log.Printf("Dropping UDP packet due to unprocessable code=%d", code)
		log.Printf("%#v", pkt[0:9])
		return
	}
	// handle boot protocol
	rxrb := "rx"
	kind := ""
	switch code {
	case RF_Pairing:
		rxrb = "rb"
		kind = "pairing"
	case RF_BootReq:
		rxrb = "rb"
		kind = "boot"
	}
	// handle packets with no source node id
	switch code {
	case RF_DataPush, RF_DataReq, RF_AckBcast, RF_BootReply, RF_Debug:
		nodeId = 0
	}
	// finally the topic
	topic := fmt.Sprintf("/rf/%d/%d/%s", groupId, nodeId, rxrb)

	// Create the payload
	payload, _ := json.Marshal(RFMessage{
		AsOf:   time.Now().UnixNano() / 1000000, // Javascript time: milliseconds
		Base64: base64.StdEncoding.EncodeToString(data),
		Kind:   kind,
	})

	// Send it off
	cc.Publish(&proto.Publish{
		Header:    proto.Header{QosLevel: proto.QosLevel(ack)},
		TopicName: topic,
		Payload:   proto.BytesPayload(payload),
	})

	// Log message, if appropriate
	if code == 9 {
		log.Printf("JeeUDP: %s", data)
	}
	log.Printf("MQTT PUB %s code=%d len=%d", topic, code, len(pkt))
}