JCS-pub/common/pkgs/storage/pool/pool.go

package pool

import (
	"sync"

	"gitlink.org.cn/cloudream/common/pkgs/async"
	clitypes "gitlink.org.cn/cloudream/jcs-pub/client/types"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/factory"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/types"
	cortypes "gitlink.org.cn/cloudream/jcs-pub/coordinator/types"
)

type userSpace struct {
	detail *clitypes.UserSpaceDetail
	store  types.ShardStore
}

func (u *userSpace) Drop() {
	if u.store != nil {
		u.store.Stop()
		u.store = nil
	}
}

type userSpaceKey struct {
	UserID      cortypes.UserID
	UserSpaceID clitypes.UserSpaceID
}

type Pool struct {
	spaces    map[userSpaceKey]*userSpace
	lock      sync.Mutex
	eventChan *types.StorageEventChan
}

func NewPool() *Pool {
	return &Pool{
		spaces:    make(map[userSpaceKey]*userSpace),
		eventChan: async.NewUnboundChannel[types.StorageEvent](),
	}
}

func (p *Pool) Drop(userID cortypes.UserID, spaceID clitypes.UserSpaceID) {
	p.lock.Lock()
	defer p.lock.Unlock()

	key := userSpaceKey{
		UserID:      userID,
		UserSpaceID: spaceID,
	}
	space := p.spaces[key]
	if space != nil {
		space.Drop()
	}

	delete(p.spaces, key)
}

func (p *Pool) GetShardStore(spaceDetail *clitypes.UserSpaceDetail) (types.ShardStore, error) {
	p.lock.Lock()
	defer p.lock.Unlock()

	key := userSpaceKey{
		UserID:      spaceDetail.UserID,
		UserSpaceID: spaceDetail.UserSpace.UserSpaceID,
	}

	space := p.spaces[key]
	if space == nil {
		space = &userSpace{
			detail: spaceDetail,
		}
		p.spaces[key] = space
	}

	if space.detail.UserSpace.Revision != spaceDetail.UserSpace.Revision {
		space.Drop()
		space.detail = spaceDetail
	}

	if space.store == nil {
		bld := factory.GetBuilder(spaceDetail)
		store, err := bld.CreateShardStore(false)
		if err != nil {
			return nil, err
		}
		space.store = store
		store.Start(p.eventChan)
	}

	return space.store, nil
}

func (p *Pool) GetBaseStore(spaceDetail *clitypes.UserSpaceDetail) (types.BaseStore, error) {
	return factory.GetBuilder(spaceDetail).CreateBaseStore(false)
}

func (p *Pool) GetMultiparter(spaceDetail *clitypes.UserSpaceDetail) (types.Multiparter, error) {
	return factory.GetBuilder(spaceDetail).CreateMultiparter(false)
}

func (p *Pool) GetS2STransfer(spaceDetail *clitypes.UserSpaceDetail) (types.S2STransfer, error) {
	return factory.GetBuilder(spaceDetail).CreateS2STransfer(false)
}

func (p *Pool) GetECMultiplier(spaceDetail *clitypes.UserSpaceDetail) (types.ECMultiplier, error) {
	return factory.GetBuilder(spaceDetail).CreateECMultiplier(false)
}