[fix]: 修改同步消息逻辑 (#236)

Co-authored-by: wenyoufu <wenyoufu@jd.com>

bot.go

@@ -2,7 +2,6 @@ package openwechat
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"io"
 	"log"
@@ -20,13 +19,14 @@ type Bot struct {
 	SyncCheckCallback   func(resp SyncCheckResponse)  // 心跳回调
 	MessageHandler      MessageHandler                // 获取消息成功的handle
 	MessageErrorHandler func(err error) bool          // 获取消息发生错误的handle, 返回true则尝试继续监听
+	Serializer          Serializer                    // 序列化器, 默认为json
+	Storage             *Storage
+	Caller              *Caller
 	once                sync.Once
 	err                 error
 	context             context.Context
 	cancel              context.CancelFunc
-	Caller              *Caller
 	self                *Self
-	Storage             *Storage
 	hotReloadStorage    HotReloadStorage
 	uuid                string
 	loginUUID           *string
@@ -51,6 +51,7 @@ func (b *Bot) Alive() bool {
 // @description: 设置设备Id
 // @receiver b
 // @param deviceId
+// TODO ADD INTO LOGIN OPTION
 func (b *Bot) SetDeviceId(deviceId string) {
 	b.deviceId = deviceId
 }
@@ -84,7 +85,7 @@ func (b *Bot) login(login BotLogin) (err error) {
 
 // Login 用户登录
 func (b *Bot) Login() error {
-	scanLogin := &SacnLogin{}
+	scanLogin := &SacnLogin{UUID: b.loginUUID}
 	return b.login(scanLogin)
 }
 
@@ -168,9 +169,10 @@ func (b *Bot) WebInit() error {
 		return err
 	}
 	// 设置当前的用户
-	b.self = &Self{bot: b, User: &resp.User}
+	b.self = &Self{bot: b, User: resp.User}
 	b.self.formatEmoji()
 	b.self.self = b.self
+	resp.ContactList.init(b.self)
 	b.Storage.Response = resp
 
 	// 通知手机客户端已经登录
@@ -221,8 +223,8 @@ func (b *Bot) syncCheck() error {
 		if !resp.Success() {
 			return resp.Err()
 		}
-		// 如果Selector不为0，则获取消息
-		if !resp.NorMal() {
+		switch resp.Selector {
+		case SelectorNewMsg:
 			messages, err := b.syncMessage()
 			if err != nil {
 				return err
@@ -235,8 +237,12 @@ func (b *Bot) syncCheck() error {
 				// 默认同步调用
 				// 如果异步调用则需自行处理
 				// 如配合 openwechat.MessageMatchDispatcher 使用
+				// NOTE: 请确保 MessageHandler 不会阻塞，否则会导致收不到后续的消息
 				b.MessageHandler(message)
 			}
+		case SelectorModContact:
+		case SelectorAddOrDelContact:
+		case SelectorModChatRoom:
 		}
 	}
 	return err
@@ -295,7 +301,7 @@ func (b *Bot) DumpTo(writer io.Writer) error {
 		WechatDomain: b.Caller.Client.Domain,
 		UUID:         b.uuid,
 	}
-	return json.NewEncoder(writer).Encode(item)
+	return b.Serializer.Encode(writer, item)
 }
 
 // IsHot returns true if is hot login otherwise false
@@ -303,7 +309,7 @@ func (b *Bot) IsHot() bool {
 	return b.hotReloadStorage != nil
 }
 
-// UUID returns current uuid of bot
+// UUID returns current UUID of bot
 func (b *Bot) UUID() string {
 	return b.uuid
 }
@@ -311,7 +317,8 @@ func (b *Bot) UUID() string {
 // SetUUID
 // @description: 设置UUID，可以用来手动登录用
 // @receiver b
-// @param uuid
+// @param UUID
+// TODO ADD INTO LOGIN OPTION
 func (b *Bot) SetUUID(uuid string) {
 	b.loginUUID = &uuid
 }
@@ -326,8 +333,7 @@ func (b *Bot) reload() error {
 		return errors.New("hotReloadStorage is nil")
 	}
 	var item HotReloadStorageItem
-	err := json.NewDecoder(b.hotReloadStorage).Decode(&item)
-	if err != nil {
+	if err := b.Serializer.Decode(b.hotReloadStorage, &item); err != nil {
 		return err
 	}
 	b.Caller.Client.SetCookieJar(item.Jar)
@@ -345,7 +351,13 @@ func NewBot(c context.Context) *Bot {
 	// 默认行为为网页版微信模式
 	caller.Client.SetMode(normal)
 	ctx, cancel := context.WithCancel(c)
-	return &Bot{Caller: caller, Storage: &Storage{}, context: ctx, cancel: cancel}
+	return &Bot{
+		Caller:     caller,
+		Storage:    &Storage{},
+		Serializer: &JsonSerializer{},
+		context:    ctx,
+		cancel:     cancel,
+	}
 }
 
 // DefaultBot 默认的Bot的构造方法,

bot_login.go

@@ -1,6 +1,7 @@
 package openwechat
 
 import (
+	"context"
 	"time"
 )
 
@@ -132,16 +133,16 @@ func NewSyncReloadDataLoginOption(duration time.Duration) BotLoginOption {
 	return &SyncReloadDataLoginOption{SyncLoopDuration: duration}
 }
 
-// WithModeOption 指定使用哪种客户端模式
-type WithModeOption struct {
+// withModeOption 指定使用哪种客户端模式
+type withModeOption struct {
 	mode Mode
 }
 
 // Prepare 实现了 BotLoginOption 接口
-func (w WithModeOption) Prepare(b *Bot) { b.Caller.Client.SetMode(w.mode) }
+func (w withModeOption) Prepare(b *Bot) { b.Caller.Client.SetMode(w.mode) }
 
 func withMode(mode Mode) BotPreparer {
-	return WithModeOption{mode: mode}
+	return withModeOption{mode: mode}
 }
 
 // btw, 这两个变量已经变了4回了, 但是为了兼容以前的代码, 还是得想着法儿让用户无感知的更新
@@ -153,6 +154,19 @@ var (
 	Desktop = withMode(desktop)
 )
 
+// WithContextOption 指定一个 context.Context 用于Bot的生命周期
+type WithContextOption struct {
+	Ctx context.Context
+}
+
+// Prepare 实现了 BotLoginOption 接口
+func (w WithContextOption) Prepare(b *Bot) {
+	if w.Ctx == nil {
+		panic("context is nil")
+	}
+	b.context, b.cancel = context.WithCancel(w.Ctx)
+}
+
 const (
 	defaultHotStorageSyncDuration = time.Minute * 5
 )
@@ -163,19 +177,21 @@ type BotLogin interface {
 }
 
 // SacnLogin 扫码登录
-type SacnLogin struct{}
+type SacnLogin struct {
+	UUID *string
+}
 
 // Login 实现了 BotLogin 接口
 func (s *SacnLogin) Login(bot *Bot) error {
 	var uuid string
-	if bot.loginUUID == nil {
+	if s.UUID == nil {
 		var err error
 		uuid, err = bot.Caller.GetLoginUUID()
 		if err != nil {
 			return err
 		}
 	} else {
-		uuid = *bot.loginUUID
+		uuid = *s.UUID
 	}
 	return s.checkLogin(bot, uuid)
 }

client.go

@@ -314,6 +314,8 @@ func (c *Client) WebWxGetHeadImg(user *User) (*http.Response, error) {
 	return c.Do(req)
 }
 
+// WebWxUploadMediaByChunk 分块上传文件
+// TODO 优化掉这个函数
 func (c *Client) WebWxUploadMediaByChunk(file *os.File, request *BaseRequest, info *LoginInfo, forUserName, toUserName string) (*http.Response, error) {
 	// 获取文件上传的类型
 	contentType, err := GetFileContentType(file)
@@ -358,7 +360,11 @@ func (c *Client) WebWxUploadMediaByChunk(file *os.File, request *BaseRequest, in
 	path.RawQuery = params.Encode()
 
 	cookies := c.Jar().Cookies(path)
-	webWxDataTicket := getWebWxDataTicket(cookies)
+
+	webWxDataTicket, err := getWebWxDataTicket(cookies)
+	if err != nil {
+		return nil, err
+	}
 
 	uploadMediaRequest := map[string]interface{}{
 		"UploadType":    2,
@@ -410,16 +416,17 @@ func (c *Client) WebWxUploadMediaByChunk(file *os.File, request *BaseRequest, in
 		return nil, err
 	}
 
+	var chunkBuff = make([]byte, chunkSize)
+
+	var formBuffer = bytes.NewBuffer(nil)
+
 	// 分块上传
 	for chunk := 0; int64(chunk) < chunks; chunk++ {
-
-		isLastTime := int64(chunk)+1 == chunks
-
 		if chunks > 1 {
 			content["chunk"] = strconv.Itoa(chunk)
 		}
 
-		var formBuffer = bytes.NewBuffer(nil)
+		formBuffer.Reset()
 
 		writer := multipart.NewWriter(formBuffer)
 
@@ -434,34 +441,33 @@ func (c *Client) WebWxUploadMediaByChunk(file *os.File, request *BaseRequest, in
 		}
 
 		w, err := writer.CreateFormFile("filename", file.Name())
-
 		if err != nil {
 			return nil, err
 		}
 
-		chunkData := make([]byte, chunkSize)
-
-		n, err := file.Read(chunkData)
+		n, err := file.Read(chunkBuff)
 
 		if err != nil && err != io.EOF {
 			return nil, err
 		}
-
-		if _, err = w.Write(chunkData[:n]); err != nil {
+		if _, err = w.Write(chunkBuff[:n]); err != nil {
 			return nil, err
 		}
-
 		ct := writer.FormDataContentType()
 		if err = writer.Close(); err != nil {
 			return nil, err
 		}
+
 		req, _ := http.NewRequest(http.MethodPost, path.String(), formBuffer)
 		req.Header.Set("Content-Type", ct)
+
 		// 发送数据
 		resp, err = c.Do(req)
 		if err != nil {
 			return nil, err
 		}
+
+		isLastTime := int64(chunk)+1 == chunks
 		// 如果不是最后一次, 解析有没有错误
 		if !isLastTime {
 			parser := MessageResponseParser{Reader: resp.Body}
@@ -591,13 +597,18 @@ func (c *Client) WebWxGetVideo(msg *Message, info *LoginInfo) (*http.Response, e
 // WebWxGetMedia 获取文件消息的文件响应
 func (c *Client) WebWxGetMedia(msg *Message, info *LoginInfo) (*http.Response, error) {
 	path, _ := url.Parse(c.Domain.FileHost() + webwxgetmedia)
+	cookies := c.Jar().Cookies(path)
+	webWxDataTicket, err := getWebWxDataTicket(cookies)
+	if err != nil {
+		return nil, err
+	}
 	params := url.Values{}
 	params.Add("sender", msg.FromUserName)
 	params.Add("mediaid", msg.MediaId)
 	params.Add("encryfilename", msg.EncryFileName)
 	params.Add("fromuser", strconv.FormatInt(info.WxUin, 10))
 	params.Add("pass_ticket", info.PassTicket)
-	params.Add("webwx_data_ticket", getWebWxDataTicket(c.Jar().Cookies(path)))
+	params.Add("webwx_data_ticket", webWxDataTicket)
 	path.RawQuery = params.Encode()
 	req, _ := http.NewRequest(http.MethodGet, path.String(), nil)
 	req.Header.Add("Referer", c.Domain.BaseHost()+"/")

cookiejar.go

@@ -9,6 +9,7 @@ import (
 )
 
 // Jar is a struct which as same as cookiejar.Jar
+// cookiejar.Jar's fields are private, so we can't use it directly
 type Jar struct {
 	PsList cookiejar.PublicSuffixList
 
@@ -57,3 +58,24 @@ type entry struct {
 	// equal Creation time. This simplifies testing.
 	seqNum uint64
 }
+
+// CookieGroup is a group of cookies
+type CookieGroup []*http.Cookie
+
+func (c CookieGroup) GetByName(cookieName string) (cookie *http.Cookie, exist bool) {
+	for _, cookie := range c {
+		if cookie.Name == cookieName {
+			return cookie, true
+		}
+	}
+	return nil, false
+}
+
+func getWebWxDataTicket(cookies []*http.Cookie) (string, error) {
+	cookieGroup := CookieGroup(cookies)
+	cookie, exist := cookieGroup.GetByName("webwx_data_ticket")
+	if !exist {
+		return "", ErrWebWxDataTicketNotFound
+	}
+	return cookie.Value, nil
+}

entity.go

@@ -57,9 +57,9 @@ type WebInitResponse struct {
 	SKey                string
 	BaseResponse        BaseResponse
 	SyncKey             SyncKey
-	User                User
-	MPSubscribeMsgList  []MPSubscribeMsg
-	ContactList         []User
+	User                *User
+	MPSubscribeMsgList  []*MPSubscribeMsg
+	ContactList         Members
 }
 
 // MPSubscribeMsg 公众号的订阅信息
@@ -68,12 +68,14 @@ type MPSubscribeMsg struct {
 	Time           int64
 	UserName       string
 	NickName       string
-	MPArticleList  []struct {
-		Title  string
-		Cover  string
-		Digest string
-		Url    string
-	}
+	MPArticleList  []*MPArticle
+}
+
+type MPArticle struct {
+	Title  string
+	Cover  string
+	Digest string
+	Url    string
 }
 
 type UserDetailItem struct {

errors.go

@@ -32,6 +32,9 @@ var (
 
 	// ErrLoginTimeout define login timeout error
 	ErrLoginTimeout = errors.New("login timeout")
+
+	// ErrWebWxDataTicketNotFound define webwx_data_ticket not found error
+	ErrWebWxDataTicketNotFound = errors.New("webwx_data_ticket not found")
 )
 
 // Error impl error interface

message_handle.go

@@ -11,14 +11,6 @@ type MessageDispatcher interface {
 	Dispatch(msg *Message)
 }
 
-// DispatchMessage 跟 MessageDispatcher 结合封装成 MessageHandler
-// Deprecated: use MessageMatchDispatcher.AsMessageHandler instead
-func DispatchMessage(dispatcher MessageDispatcher) func(msg *Message) {
-	return func(msg *Message) { dispatcher.Dispatch(msg) }
-}
-
-// MessageDispatcher impl
-
 // MessageContextHandler 消息处理函数
 type MessageContextHandler func(ctx *MessageContext)
 

parser.go

@@ -38,15 +38,6 @@ func GetRandomDeviceId() string {
 	return builder.String()
 }
 
-func getWebWxDataTicket(cookies []*http.Cookie) string {
-	for _, cookie := range cookies {
-		if cookie.Name == "webwx_data_ticket" {
-			return cookie.Value
-		}
-	}
-	return ""
-}
-
 // GetFileContentType 获取文件上传的类型
 func GetFileContentType(file multipart.File) (string, error) {
 	data := make([]byte, 512)

relations.go

@@ -10,7 +10,11 @@ type Friend struct{ *User }
 
 // implement fmt.Stringer
 func (f *Friend) String() string {
-	return fmt.Sprintf("<Friend:%s>", f.NickName)
+	display := f.NickName
+	if f.RemarkName != "" {
+		display = f.RemarkName
+	}
+	return fmt.Sprintf("<Friend:%s>", display)
 }
 
 // SetRemarkName 重命名当前好友
@@ -300,11 +304,6 @@ func (g Groups) SearchByNickName(limit int, nickName string) (results Groups) {
 	return g.Search(limit, func(group *Group) bool { return group.NickName == nickName })
 }
 
-// SearchByRemarkName 根据备注查找群组
-func (g Groups) SearchByRemarkName(limit int, remarkName string) (results Groups) {
-	return g.Search(limit, func(group *Group) bool { return group.RemarkName == remarkName })
-}
-
 // Search 根据自定义条件查找群组
 func (g Groups) Search(limit int, searchFuncList ...func(group *Group) bool) (results Groups) {
 	return g.AsMembers().Search(limit, func(user *User) bool {
@@ -445,11 +444,6 @@ func (g Groups) GetByUsername(username string) *Group {
 	return g.SearchByUserName(1, username).First()
 }
 
-// GetByRemarkName 根据remarkName查询一个Group
-func (g Groups) GetByRemarkName(remarkName string) *Group {
-	return g.SearchByRemarkName(1, remarkName).First()
-}
-
 // GetByNickName 根据nickname查询一个Group
 func (g Groups) GetByNickName(nickname string) *Group {
 	return g.SearchByNickName(1, nickname).First()

serializer.go

@@ -0,0 +1,25 @@
+package openwechat
+
+import (
+	"encoding/json"
+	"io"
+)
+
+// Serializer is an interface for encoding and decoding data.
+type Serializer interface {
+	Encode(writer io.Writer, v interface{}) error
+	Decode(reader io.Reader, v interface{}) error
+}
+
+// JsonSerializer is a serializer for json.
+type JsonSerializer struct{}
+
+// Encode encodes v to writer.
+func (j JsonSerializer) Encode(writer io.Writer, v interface{}) error {
+	return json.NewEncoder(writer).Encode(v)
+}
+
+// Decode decodes data from reader to v.
+func (j JsonSerializer) Decode(reader io.Reader, v interface{}) error {
+	return json.NewDecoder(reader).Decode(v)
+}

source/bot.md

@@ -241,7 +241,7 @@ dispatcher.OnText(func(ctx *openwechat.MessageContext){
 })
 
 // 注册消息回调函数
-bot.MessageHandler = openwechat.DispatchMessage(dispatcher)
+bot.MessageHandler = dispatcher.AsMessageHandler()
 ```
 
 `openwechat.DispatchMessage`会将消息转发给`dispatcher`对象处理

stroage.go

@@ -3,6 +3,7 @@ package openwechat
 import (
 	"io"
 	"os"
+	"sync"
 	"time"
 )
 
@@ -24,14 +25,17 @@ type HotReloadStorageItem struct {
 // HotReloadStorage 热登陆存储接口
 type HotReloadStorage io.ReadWriter
 
-// jsonFileHotReloadStorage 实现HotReloadStorage接口
-// 默认以json文件的形式存储
-type jsonFileHotReloadStorage struct {
+// fileHotReloadStorage 实现HotReloadStorage接口
+// 以文件的形式存储
+type fileHotReloadStorage struct {
 	filename string
 	file     *os.File
+	lock     sync.Mutex
 }
 
-func (j *jsonFileHotReloadStorage) Read(p []byte) (n int, err error) {
+func (j *fileHotReloadStorage) Read(p []byte) (n int, err error) {
+	j.lock.Lock()
+	defer j.lock.Unlock()
 	if j.file == nil {
 		j.file, err = os.OpenFile(j.filename, os.O_RDWR, 0600)
 		if os.IsNotExist(err) {
@@ -44,38 +48,47 @@ func (j *jsonFileHotReloadStorage) Read(p []byte) (n int, err error) {
 	return j.file.Read(p)
 }
 
-func (j *jsonFileHotReloadStorage) Write(p []byte) (n int, err error) {
+func (j *fileHotReloadStorage) Write(p []byte) (n int, err error) {
+	j.lock.Lock()
+	defer j.lock.Unlock()
 	if j.file == nil {
 		j.file, err = os.Create(j.filename)
 		if err != nil {
 			return 0, err
 		}
 	}
-	// 为什么这里要对文件进行Truncate操作呢?
-	// 这是为了方便每次Dump的时候对文件进行重新写入, 而不是追加
-	// json序列化写入只会调用一次Write方法, 所以不要把这个方法当成io.Writer的Write方法
+	// reset offset and truncate file
 	if _, err = j.file.Seek(0, io.SeekStart); err != nil {
 		return
 	}
 	if err = j.file.Truncate(0); err != nil {
 		return
 	}
+	// json decode only write once
 	return j.file.Write(p)
 }
 
-func (j *jsonFileHotReloadStorage) Close() error {
+func (j *fileHotReloadStorage) Close() error {
+	j.lock.Lock()
+	defer j.lock.Unlock()
 	if j.file == nil {
 		return nil
 	}
 	return j.file.Close()
 }
 
-// NewJsonFileHotReloadStorage 创建JsonFileHotReloadStorage
+// Deprecated: use NewFileHotReloadStorage instead
+// 不再单纯以json的格式存储，支持了用户自定义序列化方式
 func NewJsonFileHotReloadStorage(filename string) io.ReadWriteCloser {
-	return &jsonFileHotReloadStorage{filename: filename}
+	return NewFileHotReloadStorage(filename)
 }
 
-var _ HotReloadStorage = (*jsonFileHotReloadStorage)(nil)
+// NewFileHotReloadStorage implements HotReloadStorage
+func NewFileHotReloadStorage(filename string) io.ReadWriteCloser {
+	return &fileHotReloadStorage{filename: filename}
+}
+
+var _ HotReloadStorage = (*fileHotReloadStorage)(nil)
 
 type HotReloadStorageSyncer struct {
 	duration time.Duration

sync_check.go

@@ -25,7 +25,11 @@ func (s SyncCheckResponse) Success() bool {
 }
 
 func (s SyncCheckResponse) NorMal() bool {
-	return s.Success() && s.Selector == "0"
+	return s.Success() && s.Selector == SelectorNormal
+}
+
+func (s SyncCheckResponse) HasNewMessage() bool {
+	return s.Success() && s.Selector == SelectorNewMsg
 }
 
 func (s SyncCheckResponse) Err() error {

user.go

@@ -58,14 +58,14 @@ type User struct {
 // implement fmt.Stringer
 func (u *User) String() string {
 	format := "User"
-	if u.IsFriend() {
+	if u.IsSelf() {
+		format = "Self"
+	} else if u.IsFriend() {
 		format = "Friend"
 	} else if u.IsGroup() {
 		format = "Group"
 	} else if u.IsMP() {
 		format = "MP"
-	} else if u.IsSelf() {
-		format = "Self"
 	}
 	return fmt.Sprintf("<%s:%s>", format, u.NickName)
 }
@@ -288,13 +288,18 @@ func (s *Self) FileHelper() *Friend {
 	}
 	return s.fileHelper
 }
+func (s *Self) ChkFrdGrpMpNil() bool {
+	return s.friends == nil && s.groups == nil && s.mps == nil
+}
 
 // Friends 获取所有的好友
 func (s *Self) Friends(update ...bool) (Friends, error) {
-	if s.friends == nil || (len(update) > 0 && update[0]) {
+	if (len(update) > 0 && update[0]) || s.ChkFrdGrpMpNil() {
 		if _, err := s.Members(true); err != nil {
 			return nil, err
 		}
+	}
+	if s.friends == nil || (len(update) > 0 && update[0]) {
 		s.friends = s.members.Friends()
 	}
 	return s.friends, nil
@@ -302,10 +307,14 @@ func (s *Self) Friends(update ...bool) (Friends, error) {
 
 // Groups 获取所有的群组
 func (s *Self) Groups(update ...bool) (Groups, error) {
-	if s.groups == nil || (len(update) > 0 && update[0]) {
+
+	if (len(update) > 0 && update[0]) || s.ChkFrdGrpMpNil() {
 		if _, err := s.Members(true); err != nil {
 			return nil, err
 		}
+
+	}
+	if s.groups == nil || (len(update) > 0 && update[0]) {
 		s.groups = s.members.Groups()
 	}
 	return s.groups, nil
@@ -313,10 +322,12 @@ func (s *Self) Groups(update ...bool) (Groups, error) {
 
 // Mps 获取所有的公众号
 func (s *Self) Mps(update ...bool) (Mps, error) {
-	if s.mps == nil || (len(update) > 0 && update[0]) {
+	if (len(update) > 0 && update[0]) || s.ChkFrdGrpMpNil() {
 		if _, err := s.Members(true); err != nil {
 			return nil, err
 		}
+	}
+	if s.mps == nil || (len(update) > 0 && update[0]) {
 		s.mps = s.members.MPs()
 	}
 	return s.mps, nil
@@ -668,6 +679,16 @@ func (s *Self) SendVideoToGroups(video io.Reader, delay time.Duration, groups ..
 	return s.sendVideoToMembers(video, delay, members...)
 }
 
+// ContactList 获取最近的联系人列表
+func (s *Self) ContactList() Members {
+	return s.Bot().Storage.Response.ContactList
+}
+
+// MPSubscribeList 获取部分公众号文章列表
+func (s *Self) MPSubscribeList() []*MPSubscribeMsg {
+	return s.Bot().Storage.Response.MPSubscribeMsgList
+}
+
 // Members 抽象的用户组
 type Members []*User