go-pattern-examples/gomore/27_messages/message.go

package gomore

import (
	"errors"
	"time"
)

//Message for msg in Message bus
type Message struct {
	//Type 类型[code :1,2,3,4]
	Type int
	Text string
	From Session //消息来源
}

//User for user
type User struct {
	ID   uint64
	Name string
}

//Session of user
type Session struct {
	User      User
	Timestamp time.Time
}

//Subscription for user
//Subscription is a session
type Subscription struct {
	Session
	ch    chan<- Message //发送队列
	Inbox <-chan Message //接收消息的队列
}

//Publish a message to subscription queue
func (s *Subscription) Publish(msg Message) error {
	if _, ok := <-s.ch; !ok {
		return errors.New("Topic has been closed")
	}
	//用go channel 作为队列,接收消息
	s.ch <- msg

	return nil
}

//Topic that user is interested in
//Topic locate in MQ
type Topic struct {
	Name           string
	Subscribers    map[uint64]Session //user list
	MessageHistory []Message          //当前主题的消息历史,实际项目中可能需要限定大小并设置过期时间
	subscription   Subscription
}

//Queue hold all topics
type Queue struct {
	Topics map[string]*Topic //topic ID<-----> topic Object
}

//String remove Subscription
func (t *Topic) String() string {
	return t.Name
}

func (t *Topic) findSession(uid uint64) (Session, bool) {
	// Get session or create one if it's the first
	var Session session
	if t.Subscribers == nil || len(t.Subscribers) == 0 {
		return Session{}, false
	}
	if session, found := t.Subscribers[uid]; found {
		return session, true
	}
	return Session{}, false
}

func (t *Topic) addSession(uid uint64) Session {

	var Session session
	// Get session or create one if it's the first
	if session, found := t.findSession(uid); !found {
		if t.Subscribers == ni {
			t.Subscribers = make(map[uint64]Session)
		}
		session = Session{User{uid, "no name"}, time.Now()}
		t.Subscribers[uid] = session
	}
	return session
}

//Subscribe a spec topic
func (t *Topic) Subscribe(uid uint64) (Subscription, error) {

	session := t.addSession(uid)

	// Create a subscription from copy
	subscription := Subscription{session, t.subscription.ch, t.subscription.Inbox}

	return subscription, nil
}

//Unsubscribe remove Subscription
func (t *Topic) Unsubscribe(s Subscription) error {
	if t.findSession(s.User.ID) {
		delete(t.Subscribers, s.User.ID)
	}
	return nil
}

//Delete topic
func (t *Topic) Delete() error {
	t.Subscribers = nil
	t.Name = ""
	t.MessageHistory = nil
	t.subscription = Subscription{}
	return nil
}