From 9148980bd0305e6fc8c8feff1e5454659f22f661 Mon Sep 17 00:00:00 2001 From: Leonard Lyubich Date: Mon, 23 Nov 2020 15:59:06 +0300 Subject: [PATCH] [#193] services/object: Support client options in all Object services Signed-off-by: Leonard Lyubich --- pkg/services/object/head/remote.go | 7 +++++-- pkg/services/object/head/service.go | 7 +++++++ pkg/services/object/put/remote.go | 13 ++++++++++--- pkg/services/object/put/service.go | 9 +++++++++ pkg/services/object/put/streamer.go | 1 + pkg/services/object/range/remote.go | 4 +++- pkg/services/object/range/service.go | 9 +++++++++ pkg/services/object/range/streamer.go | 1 + pkg/services/object/rangehash/distributed.go | 1 + pkg/services/object/rangehash/remote.go | 4 +++- pkg/services/object/rangehash/service.go | 9 +++++++++ pkg/services/object/search/remote.go | 4 +++- pkg/services/object/search/service.go | 9 +++++++++ pkg/services/object/search/streamer.go | 1 + 14 files changed, 71 insertions(+), 8 deletions(-) diff --git a/pkg/services/object/head/remote.go b/pkg/services/object/head/remote.go index b1ed7c2b0..e6979e328 100644 --- a/pkg/services/object/head/remote.go +++ b/pkg/services/object/head/remote.go @@ -18,6 +18,8 @@ type RemoteHeader struct { keyStorage *util.KeyStorage clientCache *cache.ClientCache + + clientOpts []client.Option } // RemoteHeadPrm groups remote header operation parameters. @@ -28,10 +30,11 @@ type RemoteHeadPrm struct { } // NewRemoteHeader creates, initializes and returns new RemoteHeader instance. -func NewRemoteHeader(keyStorage *util.KeyStorage, cache *cache.ClientCache) *RemoteHeader { +func NewRemoteHeader(keyStorage *util.KeyStorage, cache *cache.ClientCache, opts ...client.Option) *RemoteHeader { return &RemoteHeader{ keyStorage: keyStorage, clientCache: cache, + clientOpts: opts, } } @@ -65,7 +68,7 @@ func (h *RemoteHeader) Head(ctx context.Context, prm *RemoteHeadPrm) (*object.Ob return nil, err } - c, err := h.clientCache.Get(key, addr) + c, err := h.clientCache.Get(key, addr, h.clientOpts...) if err != nil { return nil, errors.Wrapf(err, "(%T) could not create SDK client %s", h, addr) } diff --git a/pkg/services/object/head/service.go b/pkg/services/object/head/service.go index 02e38ade1..e0f0bab57 100644 --- a/pkg/services/object/head/service.go +++ b/pkg/services/object/head/service.go @@ -3,6 +3,7 @@ package headsvc import ( "context" + "github.com/nspcc-dev/neofs-api-go/pkg/client" objectSDK "github.com/nspcc-dev/neofs-api-go/pkg/object" "github.com/nspcc-dev/neofs-node/pkg/core/container" "github.com/nspcc-dev/neofs-node/pkg/core/netmap" @@ -152,3 +153,9 @@ func WithLogger(l *logger.Logger) Option { c.log = l } } + +func WithClientOptions(opts ...client.Option) Option { + return func(c *cfg) { + c.remoteHeader.clientOpts = opts + } +} diff --git a/pkg/services/object/put/remote.go b/pkg/services/object/put/remote.go index 88e70871e..ab57ce677 100644 --- a/pkg/services/object/put/remote.go +++ b/pkg/services/object/put/remote.go @@ -29,13 +29,18 @@ type remoteTarget struct { obj *object.Object clientCache *cache.ClientCache + + clientOpts []client.Option } // RemoteSender represents utility for // sending an object to a remote host. type RemoteSender struct { - keyStorage *util.KeyStorage + keyStorage *util.KeyStorage + clientCache *cache.ClientCache + + clientOpts []client.Option } // RemotePutPrm groups remote put operation parameters. @@ -62,7 +67,7 @@ func (t *remoteTarget) Close() (*transformer.AccessIdentifiers, error) { return nil, err } - c, err := t.clientCache.Get(key, addr) + c, err := t.clientCache.Get(key, addr, t.clientOpts...) if err != nil { return nil, errors.Wrapf(err, "(%T) could not create SDK client %s", t, addr) } @@ -83,10 +88,11 @@ func (t *remoteTarget) Close() (*transformer.AccessIdentifiers, error) { } // NewRemoteSender creates, initializes and returns new RemoteSender instance. -func NewRemoteSender(keyStorage *util.KeyStorage, cache *cache.ClientCache) *RemoteSender { +func NewRemoteSender(keyStorage *util.KeyStorage, cache *cache.ClientCache, opts ...client.Option) *RemoteSender { return &RemoteSender{ keyStorage: keyStorage, clientCache: cache, + clientOpts: opts, } } @@ -115,6 +121,7 @@ func (s *RemoteSender) PutObject(ctx context.Context, p *RemotePutPrm) error { keyStorage: s.keyStorage, addr: p.node, clientCache: s.clientCache, + clientOpts: s.clientOpts, } if err := t.WriteHeader(object.NewRawFromObject(p.obj)); err != nil { diff --git a/pkg/services/object/put/service.go b/pkg/services/object/put/service.go index 4bf954296..958f78557 100644 --- a/pkg/services/object/put/service.go +++ b/pkg/services/object/put/service.go @@ -3,6 +3,7 @@ package putsvc import ( "context" + "github.com/nspcc-dev/neofs-api-go/pkg/client" "github.com/nspcc-dev/neofs-node/pkg/core/container" "github.com/nspcc-dev/neofs-node/pkg/core/netmap" "github.com/nspcc-dev/neofs-node/pkg/core/object" @@ -53,6 +54,8 @@ type cfg struct { clientCache *cache.ClientCache log *logger.Logger + + clientOpts []client.Option } func defaultCfg() *cfg { @@ -148,3 +151,9 @@ func WithLogger(l *logger.Logger) Option { c.log = l } } + +func WithClientOptions(opts ...client.Option) Option { + return func(c *cfg) { + c.clientOpts = opts + } +} diff --git a/pkg/services/object/put/streamer.go b/pkg/services/object/put/streamer.go index 7049e0ea6..d8af83004 100644 --- a/pkg/services/object/put/streamer.go +++ b/pkg/services/object/put/streamer.go @@ -144,6 +144,7 @@ func (p *Streamer) newCommonTarget(prm *PutInitPrm) transformer.ObjectTarget { bearer: prm.common.BearerToken(), addr: addr, clientCache: p.clientCache, + clientOpts: p.clientOpts, } } }, diff --git a/pkg/services/object/range/remote.go b/pkg/services/object/range/remote.go index b9281db72..3af136e2c 100644 --- a/pkg/services/object/range/remote.go +++ b/pkg/services/object/range/remote.go @@ -29,6 +29,8 @@ type remoteRangeWriter struct { rng *object.Range clientCache *cache.ClientCache + + clientOpts []client.Option } func (r *remoteRangeWriter) WriteTo(w io.Writer) (int64, error) { @@ -42,7 +44,7 @@ func (r *remoteRangeWriter) WriteTo(w io.Writer) (int64, error) { return 0, err } - c, err := r.clientCache.Get(key, addr) + c, err := r.clientCache.Get(key, addr, r.clientOpts...) if err != nil { return 0, errors.Wrapf(err, "(%T) could not create SDK client %s", r, addr) } diff --git a/pkg/services/object/range/service.go b/pkg/services/object/range/service.go index 09c1806a4..ed12b4630 100644 --- a/pkg/services/object/range/service.go +++ b/pkg/services/object/range/service.go @@ -4,6 +4,7 @@ import ( "context" "sync" + "github.com/nspcc-dev/neofs-api-go/pkg/client" "github.com/nspcc-dev/neofs-api-go/pkg/object" "github.com/nspcc-dev/neofs-node/pkg/core/container" "github.com/nspcc-dev/neofs-node/pkg/core/netmap" @@ -42,6 +43,8 @@ type cfg struct { clientCache *cache.ClientCache log *logger.Logger + + clientOpts []client.Option } func defaultCfg() *cfg { @@ -184,3 +187,9 @@ func WithLogger(l *logger.Logger) Option { c.log = l } } + +func WithClientOptions(opts ...client.Option) Option { + return func(c *cfg) { + c.clientOpts = opts + } +} diff --git a/pkg/services/object/range/streamer.go b/pkg/services/object/range/streamer.go index 1d557877a..24a85aa13 100644 --- a/pkg/services/object/range/streamer.go +++ b/pkg/services/object/range/streamer.go @@ -186,6 +186,7 @@ loop: addr: objAddr, rng: nextRange, clientCache: p.clientCache, + clientOpts: p.clientOpts, } } diff --git a/pkg/services/object/rangehash/distributed.go b/pkg/services/object/rangehash/distributed.go index e4171680e..059656146 100644 --- a/pkg/services/object/rangehash/distributed.go +++ b/pkg/services/object/rangehash/distributed.go @@ -115,6 +115,7 @@ loop: keyStorage: h.keyStorage, node: addr, clientCache: h.clientCache, + clientOpts: h.clientOpts, } } diff --git a/pkg/services/object/rangehash/remote.go b/pkg/services/object/rangehash/remote.go index 5486a68c4..78f0cc743 100644 --- a/pkg/services/object/rangehash/remote.go +++ b/pkg/services/object/rangehash/remote.go @@ -18,6 +18,8 @@ type remoteHasher struct { node *network.Address clientCache *cache.ClientCache + + clientOpts []client.Option } func (h *remoteHasher) hashRange(ctx context.Context, prm *Prm, handler func([][]byte)) error { @@ -31,7 +33,7 @@ func (h *remoteHasher) hashRange(ctx context.Context, prm *Prm, handler func([][ return err } - c, err := h.clientCache.Get(key, addr) + c, err := h.clientCache.Get(key, addr, h.clientOpts...) if err != nil { return errors.Wrapf(err, "(%T) could not create SDK client %s", h, addr) } diff --git a/pkg/services/object/rangehash/service.go b/pkg/services/object/rangehash/service.go index 3268ab13c..35bc389e5 100644 --- a/pkg/services/object/rangehash/service.go +++ b/pkg/services/object/rangehash/service.go @@ -7,6 +7,7 @@ import ( "io" "github.com/nspcc-dev/neofs-api-go/pkg" + "github.com/nspcc-dev/neofs-api-go/pkg/client" "github.com/nspcc-dev/neofs-api-go/pkg/object" "github.com/nspcc-dev/neofs-node/pkg/core/container" "github.com/nspcc-dev/neofs-node/pkg/core/netmap" @@ -48,6 +49,8 @@ type cfg struct { clientCache *cache.ClientCache log *logger.Logger + + clientOpts []client.Option } func defaultCfg() *cfg { @@ -284,3 +287,9 @@ func WithLogger(l *logger.Logger) Option { c.log = l } } + +func WithClientOptions(opts ...client.Option) Option { + return func(c *cfg) { + c.clientOpts = opts + } +} diff --git a/pkg/services/object/search/remote.go b/pkg/services/object/search/remote.go index 14c1c5094..95fa5fb06 100644 --- a/pkg/services/object/search/remote.go +++ b/pkg/services/object/search/remote.go @@ -19,6 +19,8 @@ type remoteStream struct { addr *network.Address clientCache *cache.ClientCache + + clientOpts []client.Option } func (s *remoteStream) stream(ctx context.Context, ch chan<- []*object.ID) error { @@ -32,7 +34,7 @@ func (s *remoteStream) stream(ctx context.Context, ch chan<- []*object.ID) error return err } - c, err := s.clientCache.Get(key, addr) + c, err := s.clientCache.Get(key, addr, s.clientOpts...) if err != nil { return errors.Wrapf(err, "(%T) could not create SDK client %s", s, addr) } diff --git a/pkg/services/object/search/service.go b/pkg/services/object/search/service.go index 26617533b..0a5bd3972 100644 --- a/pkg/services/object/search/service.go +++ b/pkg/services/object/search/service.go @@ -4,6 +4,7 @@ import ( "context" "sync" + "github.com/nspcc-dev/neofs-api-go/pkg/client" "github.com/nspcc-dev/neofs-api-go/pkg/object" "github.com/nspcc-dev/neofs-node/pkg/core/container" "github.com/nspcc-dev/neofs-node/pkg/core/netmap" @@ -38,6 +39,8 @@ type cfg struct { clientCache *cache.ClientCache log *logger.Logger + + clientOpts []client.Option } func defaultCfg() *cfg { @@ -116,3 +119,9 @@ func WithLogger(l *logger.Logger) Option { c.log = l } } + +func WithClientOptions(opts ...client.Option) Option { + return func(c *cfg) { + c.clientOpts = opts + } +} diff --git a/pkg/services/object/search/streamer.go b/pkg/services/object/search/streamer.go index 7be5bdee2..472bf4213 100644 --- a/pkg/services/object/search/streamer.go +++ b/pkg/services/object/search/streamer.go @@ -165,6 +165,7 @@ loop: keyStorage: p.keyStorage, addr: addr, clientCache: p.clientCache, + clientOpts: p.clientOpts, } }