mirror of
				https://github.com/openimsdk/open-im-server.git
				synced 2025-11-04 03:13:15 +08:00 
			
		
		
		
	* refactor: refactor workflows contents. * add tool workflows. * update field. * fix: remove chat error. * Fix err. * fix error. * remove cn comment. * update workflows files. * update infra config. * move workflows. * feat: update bot. * fix: solve uncorrect outdated msg get. * update get docIDs logic. * update * update skip logic. * fix * update. * fix: delay deleteObject func. * remove unused content. * update log type. * feat: implement request batch count limit. * update * update * feat: add rocksTimeout * feat: wrap logs * feat: add logs * feat: listen config * feat: enable listen TIME_WAIT port * feat: add logs * feat: cache batch * chore: enable fullUserCache * feat: push rpc num * feat: push err * feat: with operationID * feat: sleep * feat: change 1s * feat: change log * feat: implement Getbatch in rpcCache. * feat: print getOnline cost * feat: change log * feat: change kafka and push config * feat: del interface * feat: fix err * feat: change config * feat: go mod * feat: change config * feat: change config * feat: add sleep in push * feat: warn logs * feat: logs * feat: logs * feat: change port * feat: start config * feat: remove port reuse * feat: prometheus config * feat: prometheus config * feat: prometheus config * feat: add long time send msg to grafana * feat: init * feat: init * feat: implement offline push. * feat: batch get user online * feat: implement batch Push spilt * update go mod * Revert "feat: change port" This reverts commit 06d5e944 * feat: change port * feat: change config * feat: implement kafka producer and consumer. * update format, * add PushMQ log. * feat: get all online users and init push * feat: lock in online cache * feat: config * fix: init online status * fix: add logs * fix: userIDs * fix: add logs * feat: update Handler logic. * update MQ logic. * update * update * fix: method name * fix: update OfflinePushConsumerHandler. * fix: prommetrics * fix: add logs * fix: ctx * fix: log * fix: config * feat: change port * fix: atomic online cache status --------- Co-authored-by: Monet Lee <monet_lee@163.com>
		
			
				
	
	
		
			115 lines
		
	
	
		
			2.7 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
			
		
		
	
	
			115 lines
		
	
	
		
			2.7 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
// Copyright © 2024 OpenIM. All rights reserved.
 | 
						|
//
 | 
						|
// Licensed under the Apache License, Version 2.0 (the "License");
 | 
						|
// you may not use this file except in compliance with the License.
 | 
						|
// You may obtain a copy of the License at
 | 
						|
//
 | 
						|
//     http://www.apache.org/licenses/LICENSE-2.0
 | 
						|
//
 | 
						|
// Unless required by applicable law or agreed to in writing, software
 | 
						|
// distributed under the License is distributed on an "AS IS" BASIS,
 | 
						|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 | 
						|
// See the License for the specific language governing permissions and
 | 
						|
// limitations under the License.
 | 
						|
 | 
						|
package lru
 | 
						|
 | 
						|
import (
 | 
						|
	"sync"
 | 
						|
	"time"
 | 
						|
 | 
						|
	"github.com/hashicorp/golang-lru/v2/expirable"
 | 
						|
)
 | 
						|
 | 
						|
func NewExpirationLRU[K comparable, V any](size int, successTTL, failedTTL time.Duration, target Target, onEvict EvictCallback[K, V]) LRU[K, V] {
 | 
						|
	var cb expirable.EvictCallback[K, *expirationLruItem[V]]
 | 
						|
	if onEvict != nil {
 | 
						|
		cb = func(key K, value *expirationLruItem[V]) {
 | 
						|
			onEvict(key, value.value)
 | 
						|
		}
 | 
						|
	}
 | 
						|
	core := expirable.NewLRU[K, *expirationLruItem[V]](size, cb, successTTL)
 | 
						|
	return &ExpirationLRU[K, V]{
 | 
						|
		core:       core,
 | 
						|
		successTTL: successTTL,
 | 
						|
		failedTTL:  failedTTL,
 | 
						|
		target:     target,
 | 
						|
	}
 | 
						|
}
 | 
						|
 | 
						|
type expirationLruItem[V any] struct {
 | 
						|
	lock  sync.RWMutex
 | 
						|
	err   error
 | 
						|
	value V
 | 
						|
}
 | 
						|
 | 
						|
type ExpirationLRU[K comparable, V any] struct {
 | 
						|
	lock       sync.Mutex
 | 
						|
	core       *expirable.LRU[K, *expirationLruItem[V]]
 | 
						|
	successTTL time.Duration
 | 
						|
	failedTTL  time.Duration
 | 
						|
	target     Target
 | 
						|
}
 | 
						|
 | 
						|
func (x *ExpirationLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
 | 
						|
	//TODO implement me
 | 
						|
	panic("implement me")
 | 
						|
}
 | 
						|
 | 
						|
func (x *ExpirationLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
 | 
						|
	x.lock.Lock()
 | 
						|
	v, ok := x.core.Get(key)
 | 
						|
	if ok {
 | 
						|
		x.lock.Unlock()
 | 
						|
		x.target.IncrGetSuccess()
 | 
						|
		v.lock.RLock()
 | 
						|
		defer v.lock.RUnlock()
 | 
						|
		return v.value, v.err
 | 
						|
	} else {
 | 
						|
		v = &expirationLruItem[V]{}
 | 
						|
		x.core.Add(key, v)
 | 
						|
		v.lock.Lock()
 | 
						|
		x.lock.Unlock()
 | 
						|
		defer v.lock.Unlock()
 | 
						|
		v.value, v.err = fetch()
 | 
						|
		if v.err == nil {
 | 
						|
			x.target.IncrGetSuccess()
 | 
						|
		} else {
 | 
						|
			x.target.IncrGetFailed()
 | 
						|
			x.core.Remove(key)
 | 
						|
		}
 | 
						|
		return v.value, v.err
 | 
						|
	}
 | 
						|
}
 | 
						|
 | 
						|
func (x *ExpirationLRU[K, V]) Del(key K) bool {
 | 
						|
	x.lock.Lock()
 | 
						|
	ok := x.core.Remove(key)
 | 
						|
	x.lock.Unlock()
 | 
						|
	if ok {
 | 
						|
		x.target.IncrDelHit()
 | 
						|
	} else {
 | 
						|
		x.target.IncrDelNotFound()
 | 
						|
	}
 | 
						|
	return ok
 | 
						|
}
 | 
						|
 | 
						|
func (x *ExpirationLRU[K, V]) SetHas(key K, value V) bool {
 | 
						|
	x.lock.Lock()
 | 
						|
	defer x.lock.Unlock()
 | 
						|
	if x.core.Contains(key) {
 | 
						|
		x.core.Add(key, &expirationLruItem[V]{value: value})
 | 
						|
		return true
 | 
						|
	}
 | 
						|
	return false
 | 
						|
}
 | 
						|
 | 
						|
func (x *ExpirationLRU[K, V]) Set(key K, value V) {
 | 
						|
	x.lock.Lock()
 | 
						|
	defer x.lock.Unlock()
 | 
						|
	x.core.Add(key, &expirationLruItem[V]{value: value})
 | 
						|
}
 | 
						|
 | 
						|
func (x *ExpirationLRU[K, V]) Stop() {
 | 
						|
}
 |