type Pubhub struct {

	// 自定義實(shí)現(xiàn)的map
	dataDict dict.ConcurrentDict

	// 該鎖的顆粒度太大
	//locker sync.RWMutex

	locker *locker.Locker // 自定義一個(gè)分布鎖
}

// SUBSCRIBE channel [channel ...]
func (p *Pubhub) Subscribe(c abstract.Connection, args [][]byte) protocol.Reply {
	iflen(args) < 1 {
		return protocol.NewArgNumErrReply("subscribe")
	}
	// 通道名
	keys := make([]string, 0, len(args))
	for _, arg := range args {
		keys = append(keys, string(arg))
	}
	// 加鎖
	p.locker.Locks(keys...)
	defer p.locker.Unlocks(keys...)
	for _, arg := range args {
		chanName := string(arg)
		// 記錄當(dāng)前客戶端連接訂閱的通道
		c.Subscribe(chanName)
		// 雙向鏈表，記錄通道下的客戶端連接
		var l *list.LinkedList
		raw, exist := p.dataDict.Get(chanName)
		if !exist { // 說(shuō)明該channel第一次使用
			l = list.NewLinkedList()
			p.dataDict.Put(chanName, l)
		} else {
			l, _ = raw.(*list.LinkedList)
		}
		// 未訂閱
		if !l.Contain(func(actual interface{}) bool {
			return c == actual
		}) {
			// 如果不重復(fù)，那就記錄訂閱
			logger.Debug("subscribe channel [" + chanName + "] success")
			l.Add(c)
		}
		// 回復(fù)客戶端消息
		_, err := c.Write(channelMsg(_subscribe, chanName, c.SubCount()))
		if err != nil {
			logger.Warn(err)
		}
	}
	return protocol.NewNoReply()
}

// 取消訂閱
// unsubscribes itself from all the channels using the UNSUBSCRIBE command without additional arguments
func (p *Pubhub) Unsubscribe(c abstract.Connection, args [][]byte) protocol.Reply {
	var channels []string
	iflen(args) < 1 { // 取消全部
		channels = c.GetChannels()
	} else { // 取消指定channel
		channels = make([]string, len(args))
		for i, v := range args {
			channels[i] = string(v)
		}
	}
	p.locker.Locks(channels...)
	defer p.locker.Unlocks(channels...)
	// 說(shuō)明已經(jīng)沒(méi)有訂閱的通道
	iflen(channels) == 0 {
		c.Write(noChannelMsg())
	}
	for _, channel := range channels {
		// 從客戶端中刪除當(dāng)前通道
		c.Unsubscribe(channel)
		// 獲取鏈表
		raw, ok := p.dataDict.Get(channel)
		if ok {
			// 從鏈表中刪除當(dāng)前客戶端
			l, _ := raw.(*list.LinkedList)
			l.DelAllByVal(func(actual interface{}) bool {
				return c == actual
			})
			// 如果鏈表為空，清理map
			if l.Len() == 0 {
				p.dataDict.Delete(channel)
			}
		}
		c.Write(channelMsg(_unsubscribe, channel, c.SubCount()))
	}
	return protocol.NewNoReply()
}

func (p *Pubhub) Publish(self abstract.Connection, args [][]byte) protocol.Reply {
	iflen(args) != 2 {
		return protocol.NewArgNumErrReply("publish")
	}
	channelName := string(args[0])
	// 加鎖
	p.locker.Locks(channelName)
	defer p.locker.Unlocks(channelName)
	raw, ok := p.dataDict.Get(channelName)
	if ok {
		var sendSuccess int64
		var failedClient = make(map[interface{}]struct{})
		// 取出鏈表
		l, _ := raw.(*list.LinkedList)
		// 遍歷鏈表
		l.ForEach(func(i int, val interface{}) bool {
			conn, _ := val.(abstract.Connection)
			if conn.IsClosed() {
				failedClient[val] = struct{}{}
				returntrue
			}
			if val == self { //不給自己發(fā)送
				returntrue
			}
			// 發(fā)送數(shù)據(jù)
			conn.Write(publisMsg(channelName, string(args[1])))
			sendSuccess++
			returntrue
		})
		// 剔除客戶端
		iflen(failedClient) > 0 {
			removed := l.DelAllByVal(func(actual interface{}) bool {
				_, ok := failedClient[actual]
				return ok
			})
			logger.Debugf("del %d closed client", removed)
		}
		// 返回發(fā)送的客戶端數(shù)量
		return protocol.NewIntegerReply(sendSuccess)
	}
	// 如果channel不存在
	return protocol.NewIntegerReply(0)
}

type Pubhub struct {
	// 自定義實(shí)現(xiàn)的map
	dataDict dict.ConcurrentDict
	// 該鎖的顆粒度太大
	//locker sync.RWMutex
	locker *locker.Locker // 自定義一個(gè)分布鎖
}

type Locker struct {
	mu []*sync.RWMutex
	mask uint32
}
// 順序加鎖(互斥)
func (l *Locker) Locks(keys ...string) {
	indexs := l.toLockIndex(keys...)
	for _, index := range indexs {
		mu := l.mu[index]
		mu.Lock()
	}
}
// 順序解鎖(互斥)
func (l *Locker) Unlocks(keys ...string) {
	indexs := l.toLockIndex(keys...)
	for _, index := range indexs {
		mu := l.mu[index]
		mu.Unlock()
	}
}
func (l *Locker) toLockIndex(keys ...string) []uint32 {
	// 將key轉(zhuǎn)成 切片索引[0,mask]
	mapIndex := make(map[uint32]struct{}) // 去重
	for _, key := range keys {
		mapIndex[l.spread(utils.Fnv32(key))] = struct{}{}
	}
	indices := make([]uint32, 0, len(mapIndex))
	for k := range mapIndex {
		indices = append(indices, k)
	}
	// 對(duì)索引排序
	sort.Slice(indices, func(i, j int) bool {
		return indices[i] < indices[j]
	})
	return indices
}