From 3fd6eb5e17b45e4e557cd6b156c7ddecebd8f42b Mon Sep 17 00:00:00 2001 From: Piotr Piotrowski Date: Fri, 9 Feb 2024 11:45:12 +0100 Subject: [PATCH] [ADDED] UpdateObjectStore and CreateOrUpdateObjectStore methods (#1550) Signed-off-by: Piotr Piotrowski --- jetstream/object.go | 84 +++++++++++++++++++++++++++++------ jetstream/test/object_test.go | 65 ++++++++++++++++++++++++++- 2 files changed, 134 insertions(+), 15 deletions(-) diff --git a/jetstream/object.go b/jetstream/object.go index 32715d788..b0c8a3a53 100644 --- a/jetstream/object.go +++ b/jetstream/object.go @@ -1,4 +1,4 @@ -// Copyright 2023 The NATS Authors +// Copyright 2023-2024 The NATS Authors // 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 @@ -41,6 +41,10 @@ type ( ObjectStore(ctx context.Context, bucket string) (ObjectStore, error) // CreateObjectStore will create an object store. CreateObjectStore(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error) + // UpdateObjectStore will update an existing object store. + UpdateObjectStore(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error) + // CreateOrUpdateObjectStore will create or update an object store. + CreateOrUpdateObjectStore(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error) // DeleteObjectStore will delete the underlying stream for the named object. DeleteObjectStore(ctx context.Context, bucket string) error // ObjectStoreNames is used to retrieve a list of bucket names @@ -253,10 +257,72 @@ const ( objDigestTmpl = objDigestType + "%s" ) -// CreateObjectStore will create an object store. func (js *jetStream) CreateObjectStore(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error) { + scfg, err := js.prepareObjectStoreConfig(ctx, cfg) + if err != nil { + return nil, err + } + + stream, err := js.CreateStream(ctx, scfg) + if err != nil { + if errors.Is(err, ErrStreamNameAlreadyInUse) { + // errors are joined so that backwards compatibility is retained + // and previous checks for ErrStreamNameAlreadyInUse will still work. + err = errors.Join(fmt.Errorf("%w: %s", ErrBucketExists, cfg.Bucket), err) + } + return nil, err + } + pushJS, err := js.legacyJetStream() + if err != nil { + return nil, err + } + + return mapStreamToObjectStore(js, pushJS, cfg.Bucket, stream), nil +} + +func (js *jetStream) UpdateObjectStore(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error) { + scfg, err := js.prepareObjectStoreConfig(ctx, cfg) + if err != nil { + return nil, err + } + + // Attempt to update the stream. + stream, err := js.UpdateStream(ctx, scfg) + if err != nil { + if errors.Is(err, ErrStreamNotFound) { + return nil, fmt.Errorf("%w: %s", ErrBucketNotFound, cfg.Bucket) + } + return nil, err + } + pushJS, err := js.legacyJetStream() + if err != nil { + return nil, err + } + + return mapStreamToObjectStore(js, pushJS, cfg.Bucket, stream), nil +} + +func (js *jetStream) CreateOrUpdateObjectStore(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error) { + scfg, err := js.prepareObjectStoreConfig(ctx, cfg) + if err != nil { + return nil, err + } + + stream, err := js.CreateOrUpdateStream(ctx, scfg) + if err != nil { + return nil, err + } + pushJS, err := js.legacyJetStream() + if err != nil { + return nil, err + } + + return mapStreamToObjectStore(js, pushJS, cfg.Bucket, stream), nil +} + +func (js *jetStream) prepareObjectStoreConfig(ctx context.Context, cfg ObjectStoreConfig) (StreamConfig, error) { if !validBucketRe.MatchString(cfg.Bucket) { - return nil, ErrInvalidStoreName + return StreamConfig{}, ErrInvalidStoreName } name := cfg.Bucket @@ -294,17 +360,7 @@ func (js *jetStream) CreateObjectStore(ctx context.Context, cfg ObjectStoreConfi Compression: compression, } - // Create our stream. - stream, err := js.CreateStream(ctx, scfg) - if err != nil { - return nil, err - } - pushJS, err := js.legacyJetStream() - if err != nil { - return nil, err - } - - return mapStreamToObjectStore(js, pushJS, name, stream), nil + return scfg, nil } // ObjectStore will look up and bind to an existing object store instance. diff --git a/jetstream/test/object_test.go b/jetstream/test/object_test.go index 7b9d85fc2..e315797f3 100644 --- a/jetstream/test/object_test.go +++ b/jetstream/test/object_test.go @@ -1,4 +1,4 @@ -// Copyright 2023 The NATS Authors +// Copyright 2023-2024 The NATS Authors // 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 @@ -129,6 +129,69 @@ func TestObjectBasics(t *testing.T) { expectErr(t, err, jetstream.ErrBucketNotFound) } +func TestCreateObjectStore(t *testing.T) { + s := RunBasicJetStreamServer() + defer shutdownJSServerAndRemoveStorage(t, s) + + nc, js := jsClient(t, s) + defer nc.Close() + ctx := context.Background() + + // invalid bucket name + _, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "TEST.", Description: "Test store"}) + expectErr(t, err, jetstream.ErrInvalidStoreName) + + _, err = js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "TEST", Description: "Test store"}) + expectOk(t, err) + + // Check that we can't overwrite existing bucket. + _, err = js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "TEST", Description: "New store"}) + expectErr(t, err, jetstream.ErrBucketExists) + + // assert that we're backwards compatible + expectErr(t, err, jetstream.ErrStreamNameAlreadyInUse) +} + +func TestUpdateObjectStore(t *testing.T) { + s := RunBasicJetStreamServer() + defer shutdownJSServerAndRemoveStorage(t, s) + + nc, js := jsClient(t, s) + defer nc.Close() + ctx := context.Background() + + // cannot update a non-existing bucket + _, err := js.UpdateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "TEST", Description: "Test store"}) + expectErr(t, err, jetstream.ErrBucketNotFound) + + _, err = js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "TEST", Description: "Test store"}) + expectOk(t, err) + + // update the bucket + _, err = js.UpdateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "TEST", Description: "New store"}) + expectOk(t, err) +} + +func TestCreateOrUpdateObjectStore(t *testing.T) { + s := RunBasicJetStreamServer() + defer shutdownJSServerAndRemoveStorage(t, s) + + nc, js := jsClient(t, s) + defer nc.Close() + ctx := context.Background() + + // invalid bucket name + _, err := js.CreateOrUpdateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "TEST.", Description: "Test store"}) + expectErr(t, err, jetstream.ErrInvalidStoreName) + + _, err = js.CreateOrUpdateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "TEST", Description: "Test store"}) + expectOk(t, err) + + // update the bucket + _, err = js.CreateOrUpdateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "TEST", Description: "New store"}) + expectOk(t, err) +} + func TestGetObjectDigestMismatch(t *testing.T) { s := RunBasicJetStreamServer() defer shutdownJSServerAndRemoveStorage(t, s)