package ext // XXX: There's no logging around heartbeats - how can we do this in a way that is useful // as a library? // // XXX: When we close the session because of a lost heartbeat or because of an error in the // heartbeating, there is no way to tell that; a Session will just appear to stop working import ( proto "QmfEm573cZeq3LpgccZMpngV6dXbm5gfU23F5nNUuhSxxJ/muxado/proto" "QmfEm573cZeq3LpgccZMpngV6dXbm5gfU23F5nNUuhSxxJ/muxado/proto/frame" "encoding/binary" "io" "time" ) const ( defaultHeartbeatInterval = 3 * time.Second defaultHeartbeatTolerance = 10 * time.Second ) type Heartbeat struct { sess proto.ISession accept proto.ExtAccept mark chan int interval time.Duration tolerance time.Duration } func NewDefaultHeartbeat() *Heartbeat { return NewHeartbeat(defaultHeartbeatInterval, defaultHeartbeatTolerance) } func NewHeartbeat(interval, tolerance time.Duration) *Heartbeat { return &Heartbeat{ mark: make(chan int), interval: interval, tolerance: tolerance, } } func (h *Heartbeat) Start(sess proto.ISession, accept proto.ExtAccept) frame.StreamType { h.sess = sess h.accept = accept go h.respond() go h.request() go h.check() return heartbeatExtensionType } func (h *Heartbeat) check() { t := time.NewTimer(h.interval + h.tolerance) for { select { case <-t.C: // time out waiting for a response! h.sess.Close() return case <-h.mark: t.Reset(h.interval + h.tolerance) } } } func (h *Heartbeat) respond() { // close the session on any errors defer h.sess.Close() stream, err := h.accept() if err != nil { return } // read the next heartbeat id and respond buf := make([]byte, 4) for { _, err := io.ReadFull(stream, buf) if err != nil { return } _, err = stream.Write(buf) if err != nil { return } } } func (h *Heartbeat) request() { // close the session on any errors defer h.sess.Close() // request highest possible priority for heartbeats priority := frame.StreamPriority(0x7FFFFFFF) stream, err := h.sess.OpenStream(priority, heartbeatExtensionType, false) if err != nil { return } // send heartbeats and then check that we got them back var id uint32 for { time.Sleep(h.interval) if err := binary.Write(stream, binary.BigEndian, id); err != nil { return } var respId uint32 if err := binary.Read(stream, binary.BigEndian, &respId); err != nil { return } if id != respId { return } // record the time h.mark <- 1 } }