func (u *upstreamData) runner(shutdown <-chan struct{}) {
	log := u.log
	log.Debug("starting…")
	queue := messagebus.Bus.Broadcast.Chan(messagebus.Default)
	cycleTimer := time.After(cycleInterval)
loop:
	for {
		log.Debug("waiting…")
		select {
		case <-shutdown:
			break loop
		case <-cycleTimer:
			cycleTimer = time.After(cycleInterval)
			u.RLock()
			if u.connected {
				u.RUnlock()
				remoteHeight, err := u.height()
				if nil == err {
					u.lastResponseTime = time.Now()
					u.Lock()
					u.remoteHeight = remoteHeight
					u.Unlock()
					publicKey := u.client.ServerPublicKey()
					messagebus.Bus.Announce.Send("updatetime", publicKey)
				} else {
					log.Warnf("highest block error: %s", err)
				}
				localHeight := blockheader.Height()
				digest, err := u.RemoteDigestOfHeight(localHeight)
				if nil != err {
					log.Warnf("remote digest for local height: %d  error: %s", localHeight, err)
					continue loop
				}
				u.Lock()
				u.localHeight = localHeight
				u.remoteDigestOfLocalHeight = digest
				u.Unlock()
			} else {
				u.RUnlock()
				log.Trace("upstream not connected")
			}
		case item := <-queue:
			log.Debugf("from queue: %q  %x", item.Command, item.Parameters)
			u.RLock()
			if u.connected {
				u.RUnlock()
				err := u.push(&item)
				if nil != err {
					log.Errorf("push: error: %s", err)
				}
			} else {
				u.RUnlock()
				log.Trace("upstream not connected")
			}
		}
	}
	log.Info("shutting down…")
	u.client.Close()
	log.Info("stopped")
}
loop to handle upstream communication
package upstream
import (
	"encoding/binary"
	"fmt"
	"sync"
	"time"
	zmq "github.com/pebbe/zmq4"
	"github.com/bitmark-inc/bitmarkd/announce"
	"github.com/bitmark-inc/bitmarkd/blockdigest"
	"github.com/bitmark-inc/bitmarkd/blockheader"
	"github.com/bitmark-inc/bitmarkd/counter"
	"github.com/bitmark-inc/bitmarkd/messagebus"
	"github.com/bitmark-inc/bitmarkd/mode"
	"github.com/bitmark-inc/bitmarkd/util"
	"github.com/bitmark-inc/bitmarkd/zmqutil"
	"github.com/bitmark-inc/logger"
)
type Upstream interface {
	ActiveInThePast(time.Duration) bool
	CachedRemoteDigestOfLocalHeight() blockdigest.Digest
	CachedRemoteHeight() uint64
	Connect(*util.Connection, []byte) error
	ConnectedTo() *zmqutil.Connected
	Destroy()
	GetBlockData(uint64) ([]byte, error)
	IsConnectedTo([]byte) bool
	IsConnected() bool
	LocalHeight() uint64
	Name() string
	RemoteAddr() (string, error)
	RemoteDigestOfHeight(uint64) (blockdigest.Digest, error)
	RemoteHeight() (uint64, error)
	ResetServer()
	ServerPublicKey() []byte
}
type upstreamData struct {
	Upstream
	sync.RWMutex
	log                       *logger.L
	name                      string
	client                    zmqutil.Client
	connected                 bool
	remoteHeight              uint64
	localHeight               uint64
	remoteDigestOfLocalHeight blockdigest.Digest
	shutdown                  chan<- struct{}
	lastResponseTime          time.Time
}
const (
	cycleInterval = 30 * time.Second
)
type connectedState int
const (
	stateDisconnected connectedState = iota
	stateConnected    connectedState = iota
)
var upstreamCounter counter.Counter
func New(privateKey []byte, publicKey []byte, timeout time.Duration) (Upstream, error) {
	client, event, err := zmqutil.NewClient(zmq.REQ, privateKey, publicKey, timeout, zmq.EVENT_ALL)
	if nil != err {
		return nil, err
	}
	n := upstreamCounter.Increment()
	shutdown := make(chan struct{})
	upstreamStr := fmt.Sprintf("upstream@%d", n)
	u := &upstreamData{
		name:      upstreamStr,
		log:       logger.New(upstreamStr),
		client:    client,
		connected: false,
		shutdown:  shutdown,
	}
	go u.runner(shutdown)
	go u.poller(shutdown, event)
	return u, nil
}