Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[ADDED] UpdateObjectStore and CreateOrUpdateObjectStore methods #1550

Merged
merged 2 commits into from
Feb 9, 2024
Merged

[ADDED] UpdateObjectStore and CreateOrUpdateObjectStore methods #1550

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3664988
[ADDED] UpdateObjectStore and CreateOrUpdateObjectStore methods
piotrpio Feb 8, 2024
b543342
Remove invalid comment
piotrpio Feb 9, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 71 additions & 14 deletions jetstream/object.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -253,10 +257,73 @@ 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.
// If the stream does not exist, create it.
piotrpio marked this conversation as resolved.
Show resolved Hide resolved
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
Expand Down Expand Up @@ -294,17 +361,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.
Expand Down
65 changes: 64 additions & 1 deletion jetstream/test/object_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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)
Expand Down