update: interface to uint64

Author: treeforest

example/kv/main.go

@@ -64,18 +64,19 @@ func (s *Server) Recovery(state []byte) error {
 
 // Apply 状态机执行命令的回调函数
 func (s *Server) Apply(commandName string, command []byte) {
-	cmd := s.cmdPool.Get().(*Command)
-	_ = json.Unmarshal(command, &cmd)
-
-	switch commandName {
-	case SET:
-		s.state[cmd.Key] = cmd.Value
-	case DEL:
-		delete(s.state, cmd.Key)
-	}
-
-	*cmd = Command{}
-	s.cmdPool.Put(cmd)
+	return
+	//cmd := s.cmdPool.Get().(*Command)
+	//_ = json.Unmarshal(command, &cmd)
+	//
+	//switch commandName {
+	//case SET:
+	//	s.state[cmd.Key] = cmd.Value
+	//case DEL:
+	//	delete(s.state, cmd.Key)
+	//}
+	//
+	//*cmd = Command{}
+	//s.cmdPool.Put(cmd)
 }
 
 func (s *Server) Serve(addr string) {

log.go

@@ -8,7 +8,6 @@ import (
 	"github.com/treeforest/raft/pb"
 	"io"
 	"os"
-	"strconv"
 	"sync"
 	"time"
 )
@@ -53,11 +52,11 @@ func newLog(path string, applyFunc func(string, []byte)) *Log {
 }
 
 func (l *Log) Subscribe(index uint64, ttl time.Duration) Subscription {
-	return l.ps.Subscribe(strconv.FormatUint(index, 10), ttl)
+	return l.ps.Subscribe(index, ttl)
 }
 
 func (l *Log) publish(index uint64) {
-	_ = l.ps.Publish(strconv.FormatUint(index, 10), struct{}{})
+	_ = l.ps.Publish(index, struct{}{})
 }
 
 // CommitIndex the last committed index in the log.

pubsub.go

@@ -8,11 +8,11 @@ import (
 
 type PubSub struct {
 	sync.RWMutex
-	subscriptions map[string]*Set // topic -> subscriptions
+	subscriptions map[uint64]*Set // topic -> subscriptions
 }
 
 type subscription struct {
-	top string
+	top uint64
 	ttl time.Duration
 	c   chan interface{}
 }
@@ -33,24 +33,24 @@ func (s *subscription) Listen() (interface{}, error) {
 
 func NewPubSub() *PubSub {
 	return &PubSub{
-		subscriptions: make(map[string]*Set),
+		subscriptions: make(map[uint64]*Set),
 	}
 }
 
 const (
-	subscriptionBuffSize = 50
+	subscriptionBuffSize = 500
 )
 
 // Publish 发布topic对应的item，监听topic的所有订阅者都将收到item
-func (ps *PubSub) Publish(topic string, item interface{}) error {
+func (ps *PubSub) Publish(topic uint64, item interface{}) error {
 	ps.RLock()
 	defer ps.RUnlock()
 	s, subscribed := ps.subscriptions[topic]
 	if !subscribed {
 		return errors.New("no subscribers")
 	}
 	for _, sub := range s.ToArray() {
-		c := sub.(*subscription).c
+		c := sub.c
 		// Not enough room in buffer, continue in order to not block publisher
 		if len(c) == subscriptionBuffSize {
 			continue
@@ -61,7 +61,7 @@ func (ps *PubSub) Publish(topic string, item interface{}) error {
 }
 
 // Subscribe 订阅事件，返回一个Listen接口
-func (ps *PubSub) Subscribe(topic string, ttl time.Duration) Subscription {
+func (ps *PubSub) Subscribe(topic uint64, ttl time.Duration) Subscription {
 	sub := &subscription{
 		top: topic,
 		ttl: ttl,
@@ -101,21 +101,21 @@ func (ps *PubSub) unSubscribe(sub *subscription) {
 }
 
 type Set struct {
-	items map[interface{}]struct{}
+	items map[*subscription]struct{}
 	lock  *sync.RWMutex
 }
 
 func NewSet() *Set {
-	return &Set{lock: &sync.RWMutex{}, items: make(map[interface{}]struct{})}
+	return &Set{lock: &sync.RWMutex{}, items: make(map[*subscription]struct{})}
 }
 
-func (s *Set) Add(item interface{}) {
+func (s *Set) Add(item *subscription) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 	s.items[item] = struct{}{}
 }
 
-func (s *Set) Exists(item interface{}) bool {
+func (s *Set) Exists(item *subscription) bool {
 	s.lock.RLock()
 	defer s.lock.RUnlock()
 	_, exists := s.items[item]
@@ -128,10 +128,10 @@ func (s *Set) Size() int {
 	return len(s.items)
 }
 
-func (s *Set) ToArray() []interface{} {
+func (s *Set) ToArray() []*subscription {
 	s.lock.RLock()
 	defer s.lock.RUnlock()
-	a := make([]interface{}, len(s.items))
+	a := make([]*subscription, len(s.items))
 	i := 0
 	for item := range s.items {
 		a[i] = item
@@ -143,10 +143,10 @@ func (s *Set) ToArray() []interface{} {
 func (s *Set) Clear() {
 	s.lock.Lock()
 	defer s.lock.Unlock()
-	s.items = make(map[interface{}]struct{})
+	s.items = make(map[*subscription]struct{})
 }
 
-func (s *Set) Remove(item interface{}) {
+func (s *Set) Remove(item *subscription) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 	delete(s.items, item)