code: func (p *Pubsub) work() {
  	var (
  		cnx   redis.Conn
  		read  *readPump
  		write *writePump
  	)
  
  	teardown := func() {
  		read.Close()
  		write.Close()
  		cnx.Close()
  		cnx = nil
  	}
  
  	defer teardown()
  
  	for {
  		if cnx == nil {
  			cnx = p.pool.Get()
  			read = newReadPump(cnx)
  			write = newWritePump(cnx)
  			go write.Work()
  			p.resubscribe(write)
  
  			go read.Work()
  		}
  
  		select {
  		case <-p.closer:
  			return
  		case data := <-p.send:
  			write.Data() <- data
  			data.written <- struct{}{}
  		case event := <-read.Data():
  			go p.handleEvent(event)
  		case err := <-read.Errs():
  			teardown()
  			p.errs <- err
  		case err := <-write.Errs():
  			teardown()
  			p.errs <- err
  		}
  	}
  }
docs: // Inner working loop for the emitter, runs until .Close() is called.