Merge pull request #256 from sipcapture/reconnect

fix reping

main.go

@@ -12,7 +12,7 @@ import (
 	"github.com/sipcapture/heplify/sniffer"
 )
 
-const version = "heplify 1.65.3"
+const version = "heplify 1.65.5"
 
 func createFlags() {
 

publish/file.go

@@ -16,6 +16,15 @@ func (fo *FileOutputer) Output(msg []byte) {
 	}
 }
 
+func (fo *FileOutputer) SendPingPacket(msg []byte) {
+	h, err := DecodeHEP(msg)
+	if err == nil {
+		logp.Info("%s\n", h.String())
+	} else {
+		logp.Warn("%s", err)
+	}
+}
+
 func NewFileOutputer() (*FileOutputer, error) {
 	fo := &FileOutputer{}
 	return fo, nil

publish/hep.go

@@ -21,6 +21,7 @@ type HEPOutputer struct {
 	hepQueue chan []byte
 	addr     []string
 	client   []HEPConn
+	msgPing  []byte
 }
 
 func NewHEPOutputer(serverAddr string) (*HEPOutputer, error) {
@@ -57,6 +58,7 @@ func (h *HEPOutputer) ReConnect(n int) (err error) {
 		return err
 	}
 	h.client[n].writer.Reset(h.client[n].conn)
+	h.ReSendPingPacket()
 	return err
 }
 
@@ -84,6 +86,25 @@ func (h *HEPOutputer) Output(msg []byte) {
 	h.hepQueue <- msg
 }
 
+func (h *HEPOutputer) SendPingPacket(msg []byte) {
+
+	if h.msgPing == nil {
+		h.msgPing = make([]byte, len(msg))
+	}
+
+	copy(h.msgPing, msg)
+
+	h.hepQueue <- h.msgPing
+}
+
+func (h *HEPOutputer) ReSendPingPacket() {
+
+	if h.msgPing != nil {
+		logp.Debug("collector", "send ping packet")
+		h.hepQueue <- h.msgPing
+	}
+}
+
 func (h *HEPOutputer) Send(msg []byte) {
 	for n := range h.addr {
 		h.client[n].writer.Write(msg)

publish/publisher.go

@@ -10,6 +10,7 @@ import (
 
 type Outputer interface {
 	Output(msg []byte)
+	SendPingPacket(msg []byte)
 }
 
 type Publisher struct {
@@ -36,6 +37,15 @@ func (pub *Publisher) output(msg []byte) {
 	pub.outputer.Output(msg)
 }
 
+func (pub *Publisher) setHEPPing(msg []byte) {
+	defer func() {
+		if err := recover(); err != nil {
+			logp.Err("recover setHEPPing %v", err)
+		}
+	}()
+	pub.outputer.SendPingPacket(msg)
+}
+
 func (pub *Publisher) Start(pq chan *decoder.Packet) {
 	for pkt := range pq {
 
@@ -45,12 +55,22 @@ func (pub *Publisher) Start(pq chan *decoder.Packet) {
 		if pkt.Version == 100 {
 			pub.output(pkt.Payload)
 			logp.Debug("publisher", "sent hep message from collector")
+		} else if pkt.Version == 0 {
+			//this is PING
+			msg, err := EncodeHEP(pkt)
+			if err != nil {
+				logp.Warn("%v", err)
+				continue
+			}
+			pub.setHEPPing(msg)
+			logp.Debug("publisher", "sent hep ping from collector")
 		} else {
 			msg, err := EncodeHEP(pkt)
 			if err != nil {
 				logp.Warn("%v", err)
 				continue
 			}
+
 			pub.output(msg)
 		}
 	}