Make ZMQ msg_id threadsafe to avoid ZMQ drops in ntopng

Cleanup the use of ZmqDriver.lock in two ways:
* Use defer d.lock.Unlock() like someone who actually knows golang

* Lock earlier in ZmqDriver.Send() to ensure that the msg_id in
  the header, the header send, and the message send are all done
  together.

Author: kellybyrd

transport/zmq.go

@@ -66,25 +66,27 @@ func (d *ZmqDriver) Prepare() error {
 
 func (d *ZmqDriver) Init() error {
 	d.lock.Lock()
+	defer d.lock.Unlock()
 	d.context, _ = zmq.NewContext()
 	d.publisher, _ = d.context.NewSocket(zmq.PUB)
 	if err := d.publisher.Bind(d.listenAddress); err != nil {
-		d.lock.Unlock()
 		log.Fatalf("Unable to bind: %s", err.Error())
 	}
 
 	log.Infof("Started ZMQ listener on: %s", d.listenAddress)
 
 	//  Ensure subscriber connection has time to complete
 	time.Sleep(time.Second)
-	d.lock.Unlock()
 	return nil
 }
 
 func (d *ZmqDriver) Send(key, data []byte) error {
 	var err error
 
 	msg_len := uint16(len(data))
+	// Lock before creating zmq header to ensure messageId is unique
+	d.lock.Lock()
+	defer d.lock.Unlock()
 	header := d.newZmqHeader(msg_len)
 
 	// send our header with the topic first as a multi-part message
@@ -94,26 +96,21 @@ func (d *ZmqDriver) Send(key, data []byte) error {
 		return err
 	}
 
-	d.lock.Lock()
 	bytes, err := d.publisher.SendBytes(hbytes, zmq.SNDMORE)
 	if err != nil {
 		log.Errorf("Unable to send header: %s", err.Error())
-		d.lock.Unlock()
 		return err
 	}
 	if bytes != len(hbytes) {
 		log.Errorf("Wrote the wrong number of header bytes: %d", bytes)
-		d.lock.Unlock()
 		return err
 	}
 
 	// now send the actual payload
 	if _, err = d.publisher.SendBytes(data, 0); err != nil {
 		log.Error(err)
-		d.lock.Unlock()
 		return err
 	}
-	d.lock.Unlock()
 
 	switch d.msgType {
 	case PBUF: