TrueCloudLab/frostfs-sdk-go
30
0
Fork 21
Code Issues 29 Pull requests 2 Releases Activity Actions

[#38] Refactoring

Browse source 
Signed-off-by: Denis Kirillov <denis@nspcc.ru>
This commit is contained in:
Denis Kirillov 2021-11-15 09:35:30 +03:00 committed by Alex Vanin
parent 818f38b811
commit e2c740ae8d
2 changed files with 177 additions and 245 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

411
pool/pool.go
View file

@ -19,6 +19,7 @@ import (
"github.com/nspcc-dev/neofs-sdk-go/object" "github.com/nspcc-dev/neofs-sdk-go/object"
"github.com/nspcc-dev/neofs-sdk-go/owner" "github.com/nspcc-dev/neofs-sdk-go/owner"
"github.com/nspcc-dev/neofs-sdk-go/session" "github.com/nspcc-dev/neofs-sdk-go/session"
"github.com/nspcc-dev/neofs-sdk-go/token"
"go.uber.org/zap" "go.uber.org/zap"
) )
@ -86,33 +87,33 @@ func (pb *Builder) Build(ctx context.Context, options *BuilderOptions) (Pool, er
// Pool is an interface providing connection artifacts on request. // Pool is an interface providing connection artifacts on request.
type Pool interface { type Pool interface {
client.Object Object
client.Container Container
Connection() (client.Client, *session.Token, error) Connection() (client.Client, *session.Token, error)
OwnerID() *owner.ID OwnerID() *owner.ID
WaitForContainerPresence(context.Context, *cid.ID, *ContainerPollingParams) error WaitForContainerPresence(context.Context, *cid.ID, *ContainerPollingParams) error
Close() Close()
ClientParam
} }
// ClientParam is analogue client.Object, client.Container but uses session token cache. type Object interface {
type ClientParam interface { PutObject(ctx context.Context, params *client.PutObjectParams, opts ...CallOption) (*object.ID, error)
PutObjectParam(ctx context.Context, params *client.PutObjectParams, callParam *CallParam) (*object.ID, error) DeleteObject(ctx context.Context, params *client.DeleteObjectParams, opts ...CallOption) error
DeleteObjectParam(ctx context.Context, params *client.DeleteObjectParams, callParam *CallParam) error GetObject(ctx context.Context, params *client.GetObjectParams, opts ...CallOption) (*object.Object, error)
GetObjectParam(ctx context.Context, params *client.GetObjectParams, callParam *CallParam) (*object.Object, error) GetObjectHeader(ctx context.Context, params *client.ObjectHeaderParams, opts ...CallOption) (*object.Object, error)
GetObjectHeaderParam(ctx context.Context, params *client.ObjectHeaderParams, callParam *CallParam) (*object.Object, error) ObjectPayloadRangeData(ctx context.Context, params *client.RangeDataParams, opts ...CallOption) ([]byte, error)
ObjectPayloadRangeDataParam(ctx context.Context, params *client.RangeDataParams, callParam *CallParam) ([]byte, error) ObjectPayloadRangeSHA256(ctx context.Context, params *client.RangeChecksumParams, opts ...CallOption) ([][32]byte, error)
ObjectPayloadRangeSHA256Param(ctx context.Context, params *client.RangeChecksumParams, callParam *CallParam) ([][32]byte, error) ObjectPayloadRangeTZ(ctx context.Context, params *client.RangeChecksumParams, opts ...CallOption) ([][64]byte, error)
ObjectPayloadRangeTZParam(ctx context.Context, params *client.RangeChecksumParams, callParam *CallParam) ([][64]byte, error) SearchObject(ctx context.Context, params *client.SearchObjectParams, opts ...CallOption) ([]*object.ID, error)
SearchObjectParam(ctx context.Context, params *client.SearchObjectParams, callParam *CallParam) ([]*object.ID, error) }
PutContainerParam(ctx context.Context, cnr *container.Container, callParam *CallParam) (*cid.ID, error)
GetContainerParam(ctx context.Context, cid *cid.ID, callParam *CallParam) (*container.Container, error) type Container interface {
ListContainersParam(ctx context.Context, ownerID *owner.ID, callParam *CallParam) ([]*cid.ID, error) PutContainer(ctx context.Context, cnr *container.Container, opts ...CallOption) (*cid.ID, error)
DeleteContainerParam(ctx context.Context, cid *cid.ID, callParam *CallParam) error GetContainer(ctx context.Context, cid *cid.ID, opts ...CallOption) (*container.Container, error)
GetEACLParam(ctx context.Context, cid *cid.ID, callParam *CallParam) (*client.EACLWithSignature, error) ListContainers(ctx context.Context, ownerID *owner.ID, opts ...CallOption) ([]*cid.ID, error)
SetEACLParam(ctx context.Context, table *eacl.Table, callParam *CallParam) error DeleteContainer(ctx context.Context, cid *cid.ID, opts ...CallOption) error
AnnounceContainerUsedSpaceParam(ctx context.Context, announce []container.UsedSpaceAnnouncement, callParam *CallParam) error GetEACL(ctx context.Context, cid *cid.ID, opts ...CallOption) (*client.EACLWithSignature, error)
SetEACL(ctx context.Context, table *eacl.Table, opts ...CallOption) error
AnnounceContainerUsedSpace(ctx context.Context, announce []container.UsedSpaceAnnouncement, opts ...CallOption) error
} }
type clientPack struct { type clientPack struct {
@ -121,11 +122,46 @@ type clientPack struct {
address string address string
} }
type CallParam struct { type CallOption func(config *callConfig)
isRetry bool
Key *ecdsa.PrivateKey
Options []client.CallOption type callConfig struct {
isRetry bool
key *ecdsa.PrivateKey
btoken *token.BearerToken
stoken *session.Token
}
func WithKey(key *ecdsa.PrivateKey) CallOption {
return func(config *callConfig) {
config.key = key
}
}
func WithBearer(token *token.BearerToken) CallOption {
return func(config *callConfig) {
config.btoken = token
}
}
func WithSession(token *session.Token) CallOption {
return func(config *callConfig) {
config.stoken = token
}
}
func retry() CallOption {
return func(config *callConfig) {
config.isRetry = true
}
}
func cfgFromOpts(opts ...CallOption) *callConfig {
var cfg = new(callConfig)
for _, opt := range opts {
opt(cfg)
}
return cfg
} }
var _ Pool = (*pool)(nil) var _ Pool = (*pool)(nil)
@ -315,162 +351,44 @@ func (p *pool) OwnerID() *owner.ID {
return p.owner return p.owner
} }
func (p *pool) conn(option []client.CallOption) (client.Client, []client.CallOption, error) {
conn, token, err := p.Connection()
if err != nil {
return nil, nil, err
}
return conn, append([]client.CallOption{client.WithSession(token)}, option...), nil
}
func formCacheKey(address string, key *ecdsa.PrivateKey) string { func formCacheKey(address string, key *ecdsa.PrivateKey) string {
k := keys.PrivateKey{PrivateKey: *key} k := keys.PrivateKey{PrivateKey: *key}
return address + k.String() return address + k.String()
} }
func (p *pool) connParam(ctx context.Context, param *CallParam) (*clientPack, []client.CallOption, error) { func (p *pool) conn(ctx context.Context, cfg *callConfig) (*clientPack, []client.CallOption, error) {
cp, err := p.connection() cp, err := p.connection()
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
key := p.key clientCallOptions := make([]client.CallOption, 0, 3)
if param.Key != nil {
key = param.Key
}
param.Options = append(param.Options, client.WithKey(key)) key := p.key
if cfg.key != nil {
key = cfg.key
}
clientCallOptions = append(clientCallOptions, client.WithKey(key))
sessionToken := cfg.stoken
if sessionToken == nil {
cacheKey := formCacheKey(cp.address, key) cacheKey := formCacheKey(cp.address, key)
token := p.cache.Get(cacheKey) sessionToken = p.cache.Get(cacheKey)
if token == nil { if sessionToken == nil {
token, err = cp.client.CreateSession(ctx, math.MaxUint32, param.Options...) sessionToken, err = cp.client.CreateSession(ctx, math.MaxUint32, clientCallOptions...)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
_ = p.cache.Put(cacheKey, token) _ = p.cache.Put(cacheKey, sessionToken)
}
}
clientCallOptions = append(clientCallOptions, client.WithSession(sessionToken))
if cfg.btoken != nil {
clientCallOptions = append(clientCallOptions, client.WithBearer(cfg.btoken))
} }
return cp, append([]client.CallOption{client.WithSession(token)}, param.Options...), nil return cp, clientCallOptions, nil
}
func (p *pool) PutObject(ctx context.Context, params *client.PutObjectParams, option ...client.CallOption) (*object.ID, error) {
conn, options, err := p.conn(option)
if err != nil {
return nil, err
}
return conn.PutObject(ctx, params, options...)
}
func (p *pool) DeleteObject(ctx context.Context, params *client.DeleteObjectParams, option ...client.CallOption) error {
conn, options, err := p.conn(option)
if err != nil {
return err
}
return conn.DeleteObject(ctx, params, options...)
}
func (p *pool) GetObject(ctx context.Context, params *client.GetObjectParams, option ...client.CallOption) (*object.Object, error) {
conn, options, err := p.conn(option)
if err != nil {
return nil, err
}
return conn.GetObject(ctx, params, options...)
}
func (p *pool) GetObjectHeader(ctx context.Context, params *client.ObjectHeaderParams, option ...client.CallOption) (*object.Object, error) {
conn, options, err := p.conn(option)
if err != nil {
return nil, err
}
return conn.GetObjectHeader(ctx, params, options...)
}
func (p *pool) ObjectPayloadRangeData(ctx context.Context, params *client.RangeDataParams, option ...client.CallOption) ([]byte, error) {
conn, options, err := p.conn(option)
if err != nil {
return nil, err
}
return conn.ObjectPayloadRangeData(ctx, params, options...)
}
func (p *pool) ObjectPayloadRangeSHA256(ctx context.Context, params *client.RangeChecksumParams, option ...client.CallOption) ([][32]byte, error) {
conn, options, err := p.conn(option)
if err != nil {
return nil, err
}
return conn.ObjectPayloadRangeSHA256(ctx, params, options...)
}
func (p *pool) ObjectPayloadRangeTZ(ctx context.Context, params *client.RangeChecksumParams, option ...client.CallOption) ([][64]byte, error) {
conn, options, err := p.conn(option)
if err != nil {
return nil, err
}
return conn.ObjectPayloadRangeTZ(ctx, params, options...)
}
func (p *pool) SearchObject(ctx context.Context, params *client.SearchObjectParams, option ...client.CallOption) ([]*object.ID, error) {
conn, options, err := p.conn(option)
if err != nil {
return nil, err
}
return conn.SearchObject(ctx, params, options...)
}
func (p *pool) PutContainer(ctx context.Context, cnr *container.Container, option ...client.CallOption) (*cid.ID, error) {
conn, options, err := p.conn(option)
if err != nil {
return nil, err
}
return conn.PutContainer(ctx, cnr, options...)
}
func (p *pool) GetContainer(ctx context.Context, cid *cid.ID, option ...client.CallOption) (*container.Container, error) {
conn, options, err := p.conn(option)
if err != nil {
return nil, err
}
return conn.GetContainer(ctx, cid, options...)
}
func (p *pool) ListContainers(ctx context.Context, ownerID *owner.ID, option ...client.CallOption) ([]*cid.ID, error) {
conn, options, err := p.conn(option)
if err != nil {
return nil, err
}
return conn.ListContainers(ctx, ownerID, options...)
}
func (p *pool) DeleteContainer(ctx context.Context, cid *cid.ID, option ...client.CallOption) error {
conn, options, err := p.conn(option)
if err != nil {
return err
}
return conn.DeleteContainer(ctx, cid, options...)
}
func (p *pool) GetEACL(ctx context.Context, cid *cid.ID, option ...client.CallOption) (*client.EACLWithSignature, error) {
conn, options, err := p.conn(option)
if err != nil {
return nil, err
}
return conn.GetEACL(ctx, cid, options...)
}
func (p *pool) SetEACL(ctx context.Context, table *eacl.Table, option ...client.CallOption) error {
conn, options, err := p.conn(option)
if err != nil {
return err
}
return conn.SetEACL(ctx, table, options...)
}
func (p *pool) AnnounceContainerUsedSpace(ctx context.Context, announce []container.UsedSpaceAnnouncement, option ...client.CallOption) error {
conn, options, err := p.conn(option)
if err != nil {
return err
}
return conn.AnnounceContainerUsedSpace(ctx, announce, options...)
} }
func (p *pool) checkSessionTokenErr(err error, address string) bool { func (p *pool) checkSessionTokenErr(err error, address string) bool {
@ -486,197 +404,212 @@ func (p *pool) checkSessionTokenErr(err error, address string) bool {
return false return false
} }
func (p *pool) PutObjectParam(ctx context.Context, params *client.PutObjectParams, callParam *CallParam) (*object.ID, error) { func (p *pool) PutObject(ctx context.Context, params *client.PutObjectParams, opts ...CallOption) (*object.ID, error) {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
res, err := cp.client.PutObject(ctx, params, options...) res, err := cp.client.PutObject(ctx, params, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.PutObjectParam(ctx, params, callParam) return p.PutObject(ctx, params, opts...)
} }
return res, err return res, err
} }
func (p *pool) DeleteObjectParam(ctx context.Context, params *client.DeleteObjectParams, callParam *CallParam) error { func (p *pool) DeleteObject(ctx context.Context, params *client.DeleteObjectParams, opts ...CallOption) error {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return err return err
} }
err = cp.client.DeleteObject(ctx, params, options...) err = cp.client.DeleteObject(ctx, params, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.DeleteObjectParam(ctx, params, callParam) return p.DeleteObject(ctx, params, opts...)
} }
return err return err
} }
func (p *pool) GetObjectParam(ctx context.Context, params *client.GetObjectParams, callParam *CallParam) (*object.Object, error) { func (p *pool) GetObject(ctx context.Context, params *client.GetObjectParams, opts ...CallOption) (*object.Object, error) {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
res, err := cp.client.GetObject(ctx, params, options...) res, err := cp.client.GetObject(ctx, params, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.GetObjectParam(ctx, params, callParam) return p.GetObject(ctx, params, opts...)
} }
return res, err return res, err
} }
func (p *pool) GetObjectHeaderParam(ctx context.Context, params *client.ObjectHeaderParams, callParam *CallParam) (*object.Object, error) { func (p *pool) GetObjectHeader(ctx context.Context, params *client.ObjectHeaderParams, opts ...CallOption) (*object.Object, error) {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
res, err := cp.client.GetObjectHeader(ctx, params, options...) res, err := cp.client.GetObjectHeader(ctx, params, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.GetObjectHeaderParam(ctx, params, callParam) return p.GetObjectHeader(ctx, params, opts...)
} }
return res, err return res, err
} }
func (p *pool) ObjectPayloadRangeDataParam(ctx context.Context, params *client.RangeDataParams, callParam *CallParam) ([]byte, error) { func (p *pool) ObjectPayloadRangeData(ctx context.Context, params *client.RangeDataParams, opts ...CallOption) ([]byte, error) {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
res, err := cp.client.ObjectPayloadRangeData(ctx, params, options...) res, err := cp.client.ObjectPayloadRangeData(ctx, params, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.ObjectPayloadRangeDataParam(ctx, params, callParam) return p.ObjectPayloadRangeData(ctx, params, opts...)
} }
return res, err return res, err
} }
func (p *pool) ObjectPayloadRangeSHA256Param(ctx context.Context, params *client.RangeChecksumParams, callParam *CallParam) ([][32]byte, error) { func (p *pool) ObjectPayloadRangeSHA256(ctx context.Context, params *client.RangeChecksumParams, opts ...CallOption) ([][32]byte, error) {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
res, err := cp.client.ObjectPayloadRangeSHA256(ctx, params, options...) res, err := cp.client.ObjectPayloadRangeSHA256(ctx, params, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.ObjectPayloadRangeSHA256Param(ctx, params, callParam) return p.ObjectPayloadRangeSHA256(ctx, params, opts...)
} }
return res, err return res, err
} }
func (p *pool) ObjectPayloadRangeTZParam(ctx context.Context, params *client.RangeChecksumParams, callParam *CallParam) ([][64]byte, error) { func (p *pool) ObjectPayloadRangeTZ(ctx context.Context, params *client.RangeChecksumParams, opts ...CallOption) ([][64]byte, error) {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
res, err := cp.client.ObjectPayloadRangeTZ(ctx, params, options...) res, err := cp.client.ObjectPayloadRangeTZ(ctx, params, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.ObjectPayloadRangeTZParam(ctx, params, callParam) return p.ObjectPayloadRangeTZ(ctx, params, opts...)
} }
return res, err return res, err
} }
func (p *pool) SearchObjectParam(ctx context.Context, params *client.SearchObjectParams, callParam *CallParam) ([]*object.ID, error) { func (p *pool) SearchObject(ctx context.Context, params *client.SearchObjectParams, opts ...CallOption) ([]*object.ID, error) {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
res, err := cp.client.SearchObject(ctx, params, options...) res, err := cp.client.SearchObject(ctx, params, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.SearchObjectParam(ctx, params, callParam) return p.SearchObject(ctx, params, opts...)
} }
return res, err return res, err
} }
func (p *pool) PutContainerParam(ctx context.Context, cnr *container.Container, callParam *CallParam) (*cid.ID, error) { func (p *pool) PutContainer(ctx context.Context, cnr *container.Container, opts ...CallOption) (*cid.ID, error) {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
res, err := cp.client.PutContainer(ctx, cnr, options...) res, err := cp.client.PutContainer(ctx, cnr, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.PutContainerParam(ctx, cnr, callParam) return p.PutContainer(ctx, cnr, opts...)
} }
return res, err return res, err
} }
func (p *pool) GetContainerParam(ctx context.Context, cid *cid.ID, callParam *CallParam) (*container.Container, error) { func (p *pool) GetContainer(ctx context.Context, cid *cid.ID, opts ...CallOption) (*container.Container, error) {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
res, err := cp.client.GetContainer(ctx, cid, options...) res, err := cp.client.GetContainer(ctx, cid, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.GetContainerParam(ctx, cid, callParam) return p.GetContainer(ctx, cid, opts...)
} }
return res, err return res, err
} }
func (p *pool) ListContainersParam(ctx context.Context, ownerID *owner.ID, callParam *CallParam) ([]*cid.ID, error) { func (p *pool) ListContainers(ctx context.Context, ownerID *owner.ID, opts ...CallOption) ([]*cid.ID, error) {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
res, err := cp.client.ListContainers(ctx, ownerID, options...) res, err := cp.client.ListContainers(ctx, ownerID, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.ListContainersParam(ctx, ownerID, callParam) return p.ListContainers(ctx, ownerID, opts...)
} }
return res, err return res, err
} }
func (p *pool) DeleteContainerParam(ctx context.Context, cid *cid.ID, callParam *CallParam) error { func (p *pool) DeleteContainer(ctx context.Context, cid *cid.ID, opts ...CallOption) error {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return err return err
} }
err = cp.client.DeleteContainer(ctx, cid, options...) err = cp.client.DeleteContainer(ctx, cid, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.DeleteContainerParam(ctx, cid, callParam) return p.DeleteContainer(ctx, cid, opts...)
} }
return err return err
} }
func (p *pool) GetEACLParam(ctx context.Context, cid *cid.ID, callParam *CallParam) (*client.EACLWithSignature, error) { func (p *pool) GetEACL(ctx context.Context, cid *cid.ID, opts ...CallOption) (*client.EACLWithSignature, error) {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
res, err := cp.client.GetEACL(ctx, cid, options...) res, err := cp.client.GetEACL(ctx, cid, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.GetEACLParam(ctx, cid, callParam) return p.GetEACL(ctx, cid, opts...)
} }
return res, err return res, err
} }
func (p *pool) SetEACLParam(ctx context.Context, table *eacl.Table, callParam *CallParam) error { func (p *pool) SetEACL(ctx context.Context, table *eacl.Table, opts ...CallOption) error {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return err return err
} }
err = cp.client.SetEACL(ctx, table, options...) err = cp.client.SetEACL(ctx, table, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.SetEACLParam(ctx, table, callParam) return p.SetEACL(ctx, table, opts...)
} }
return err return err
} }
func (p *pool) AnnounceContainerUsedSpaceParam(ctx context.Context, announce []container.UsedSpaceAnnouncement, callParam *CallParam) error { func (p *pool) AnnounceContainerUsedSpace(ctx context.Context, announce []container.UsedSpaceAnnouncement, opts ...CallOption) error {
cp, options, err := p.connParam(ctx, callParam) cfg := cfgFromOpts(opts...)
cp, options, err := p.conn(ctx, cfg)
if err != nil { if err != nil {
return err return err
} }
err = cp.client.AnnounceContainerUsedSpace(ctx, announce, options...) err = cp.client.AnnounceContainerUsedSpace(ctx, announce, options...)
if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry { if p.checkSessionTokenErr(err, cp.address) && !cfg.isRetry {
callParam.isRetry = true opts = append(opts, retry())
return p.AnnounceContainerUsedSpaceParam(ctx, announce, callParam) return p.AnnounceContainerUsedSpace(ctx, announce, opts...)
} }
return err return err
} }

9
pool/pool_test.go
View file

@ -323,8 +323,7 @@ func TestSessionCache(t *testing.T) {
}).MaxTimes(3) }).MaxTimes(3)
mockClient.EXPECT().GetObject(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("session token does not exist")) mockClient.EXPECT().GetObject(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("session token does not exist"))
mockClient.EXPECT().PutObject(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("session token does not exist")) mockClient.EXPECT().PutObject(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil)
mockClient.EXPECT().PutObject(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil)
return mockClient, nil return mockClient, nil
} }
@ -352,7 +351,7 @@ func TestSessionCache(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
require.Contains(t, tokens, st) require.Contains(t, tokens, st)
_, err = pool.GetObjectParam(ctx, nil, &CallParam{isRetry: true}) _, err = pool.GetObject(ctx, nil, retry())
require.Error(t, err) require.Error(t, err)
// cache must not contain session token // cache must not contain session token
@ -360,7 +359,7 @@ func TestSessionCache(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
require.Nil(t, st) require.Nil(t, st)
_, err = pool.PutObjectParam(ctx, nil, &CallParam{}) _, err = pool.PutObject(ctx, nil)
require.NoError(t, err) require.NoError(t, err)
// cache must contain session token // cache must contain session token
@ -413,7 +412,7 @@ func TestSessionCacheWithKey(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
require.Contains(t, tokens, st) require.Contains(t, tokens, st)
_, err = pool.GetObjectParam(ctx, nil, &CallParam{Key: &key2.PrivateKey}) _, err = pool.GetObject(ctx, nil, WithKey(&key2.PrivateKey))
require.NoError(t, err) require.NoError(t, err)
require.Len(t, tokens, 2) require.Len(t, tokens, 2)
} }