func (s *server) Subscribe(p *Chat_Consumer, stream ChatService_SubscribeServer) error {
	ep := s.read_ep(p.Id)
	if ep == nil {
		log.Errorf("cannot find endpoint %v", p)
		return ERROR_NOT_EXISTS
	}
	consumerid := atomic.AddUint64(&s.consumerid_autoinc, 1)
	e := make(chan error, 1)
	ep.mu.Lock()
	if p.From == -1 {
		p.From = ep.StartOffset + int64(len(ep.Inbox))
	}
	ep.consumers[consumerid] = &Consumer{p.From, func(msg *Chat_Message) {
		if err := stream.Send(msg); err != nil {
			select {
			case e <- err:
			default:
			}
		}
	}}
	ep.mu.Unlock()
	defer func() {
		ep.mu.Lock()
		delete(ep.consumers, consumerid)
		ep.mu.Unlock()
	}()
	ep.notifyConsumers()
	select {
	case <-stream.Context().Done():
	case err := <-e:
		return err
	}
	return nil
}
subscribe to an endpoint & receive server streams
package main
import (
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"strconv"
	"sync"
	"sync/atomic"
	"syscall"
	"time"
	"github.com/xtaci/chat/kafka"
	"github.com/Shopify/sarama"
	log "github.com/Sirupsen/logrus"
	"github.com/boltdb/bolt"
	"golang.org/x/net/context"
	. "github.com/xtaci/chat/proto"
	cli "gopkg.in/urfave/cli.v2"
)
var (
	OK                   = &Chat_Nil{}
	ERROR_ALREADY_EXISTS = errors.New("id already exists")
	ERROR_NOT_EXISTS     = errors.New("id not exists")
)
type Consumer struct {
	offset   int64  
	pushFunc func(msg *Chat_Message)
}
type EndPoint struct {
	retention   int
	StartOffset int64  
	Inbox       []Chat_Message
	consumers   map[uint64]*Consumer
	chReady     chan struct{}
	die         chan struct{}
	mu          sync.Mutex
}
func newEndPoint(retention int) *EndPoint {
	ep := &EndPoint{}
	ep.retention = retention
	ep.chReady = make(chan struct{}, 1)
	ep.consumers = make(map[uint64]*Consumer)
	ep.StartOffset = 1
	ep.die = make(chan struct{})
	go ep.pushTask()
	return ep
}
func (ep *EndPoint) push(msg *Chat_Message) {
	if len(ep.Inbox) > ep.retention {
		ep.Inbox = append(ep.Inbox[1:], *msg)
		ep.StartOffset++
	} else {
		ep.Inbox = append(ep.Inbox, *msg)
	}
	ep.notifyConsumers()
}
func (ep *EndPoint) close() {
	close(ep.die)
}
func (ep *EndPoint) notifyConsumers() {
	select {
	case ep.chReady <- struct{}{}:
	default:
	}
}
func (ep *EndPoint) pushTask() {
	for {
		select {
		case <-ep.chReady:
			ep.mu.Lock()
			for _, consumer := range ep.consumers {
				idx := consumer.offset - ep.StartOffset
				if idx < 0 {  
					idx = 0
				}
				for i := idx; i < int64(len(ep.Inbox)); i++ {
					ep.Inbox[i].Offset = i + ep.StartOffset
					consumer.pushFunc(&ep.Inbox[i])
				}
				consumer.offset = ep.StartOffset + int64(len(ep.Inbox))
			}
			ep.mu.Unlock()
		case <-ep.die:
		}
	}
}
type server struct {
	consumerid_autoinc uint64
	kafkaOffset        int64
	offsetBucket       string
	retention          int
	boltdb             string
	bucket             string
	interval           time.Duration
	eps                map[uint64]*EndPoint  
	sync.RWMutex
}
func (s *server) init(c *cli.Context) {
	s.retention = c.Int("retention")
	s.boltdb = c.String("boltdb")
	s.bucket = c.String("bucket")
	s.interval = c.Duration("write-interval")
	s.offsetBucket = c.String("kafka-bucket")
	s.eps = make(map[uint64]*EndPoint)
	s.restore()
	go s.receive()
	go s.persistence_task()
}
func (s *server) read_ep(id uint64) *EndPoint {
	s.RLock()
	defer s.RUnlock()
	return s.eps[id]
}