pattern for a heartbeat that checks on a mgo session

recovering_session.go

// RecoveringSession is a session that has methods of ensuring it is still
// alive.
type RecoveringSession struct {
  *mgo.Session
}

// EnsureAlive attempts to ping the session, and refreshes it if something has
// gone wrong.
func (s *RecoveringSession) EnsureAlive() {
  // Squelch mgo panicing over a closed session
  defer func() { recover() }()

  if err := s.Ping(); err != nil {
    s.Refresh()
  }
}

// Monitor ensures a session is alive at a given interval. When done monitoring,
// close the returned channel.
func (s *RecoveringSession) Monitor(d time.Duration) chan<- struct{} {
  c := make(chan struct{})

  go func(stop <-chan struct{}) {
    ticker := time.NewTicker(d)
    defer ticker.Stop()

    for {
      select {
        case <-ticker.C:
          s.EnsureAlive()
        case <-stop:
          return
      }
    }
  }(c)

  return c
}

// Copy the underlying session and create a new RecoveringSession
func (s *RecoveringSession) Copy() *RecoveringSession {
  return NewRecoveringSession(s.Session.Copy())
}

// Clone the underlying session and create a new RecoveringSession
func (s *RecoveringSession) Clone() *RecoveringSession {
  return NewRecoveringSession(s.Session.Clone())
}

// NewRecoveringSession creates a new RecoveringSession
func NewRecoveringSession(s *mgo.Session) *RecoveringSession {
  return &RecoveringSession{s}
}

// Usage:
s := NewRecoveringSession(mgo.Dial(*dburl))
defer s.Close()
defer close(s.Monitor(time.Second * 30))