-
Notifications
You must be signed in to change notification settings - Fork 665
/
object.go
1607 lines (1387 loc) · 46.1 KB
/
object.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// 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
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package jetstream
import (
"bytes"
"context"
"crypto/sha256"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"hash"
"io"
"net"
"os"
"strings"
"sync"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/internal/parser"
"github.com/nats-io/nuid"
)
type (
// ObjectStoreManager is used to manage object stores. It provides methods
// CRUD operations on object stores.
ObjectStoreManager interface {
// ObjectStore will look up and bind to an existing object store
// instance.
//
// If the object store with given name does not exist, ErrBucketNotFound
// will be returned.
ObjectStore(ctx context.Context, bucket string) (ObjectStore, error)
// CreateObjectStore will create a new object store with the given
// configuration.
//
// If the object store with given name already exists, ErrBucketExists
// will be returned.
CreateObjectStore(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error)
// UpdateObjectStore will update an existing object store with the given
// configuration.
//
// If the object store with given name does not exist, ErrBucketNotFound
// will be returned.
UpdateObjectStore(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error)
// CreateOrUpdateObjectStore will create a new object store with the given
// configuration if it does not exist, or update an existing object store
// with the given configuration.
CreateOrUpdateObjectStore(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error)
// DeleteObjectStore will delete the provided object store.
//
// If the object store with given name does not exist, ErrBucketNotFound
// will be returned.
DeleteObjectStore(ctx context.Context, bucket string) error
// ObjectStoreNames is used to retrieve a list of bucket names.
// It returns an ObjectStoreNamesLister exposing a channel to receive
// the names of the object stores.
//
// The lister will always close the channel when done (either all names
// have been read or an error occurred) and therefore can be used in a
// for-range loop.
ObjectStoreNames(ctx context.Context) ObjectStoreNamesLister
// ObjectStores is used to retrieve a list of bucket statuses.
// It returns an ObjectStoresLister exposing a channel to receive
// the statuses of the object stores.
//
// The lister will always close the channel when done (either all statuses
// have been read or an error occurred) and therefore can be used in a
// for-range loop.
ObjectStores(ctx context.Context) ObjectStoresLister
}
// ObjectStore contains methods to operate on an object store.
// Using the ObjectStore interface, it is possible to:
//
// - Perform CRUD operations on objects (Get, Put, Delete).
// Get and put expose convenience methods to work with
// byte slices, strings and files, in addition to streaming [io.Reader]
// - Get information about an object without retrieving it.
// - Update the metadata of an object.
// - Add links to other objects or object stores.
// - Watch for updates to a store
// - List information about objects in a store
// - Retrieve status and configuration of an object store.
ObjectStore interface {
// Put will place the contents from the reader into a new object. If the
// object already exists, it will be overwritten. The object name is
// required and is taken from the ObjectMeta.Name field.
//
// The reader will be read until EOF. ObjectInfo will be returned, containing
// the object's metadata, digest and instance information.
Put(ctx context.Context, obj ObjectMeta, reader io.Reader) (*ObjectInfo, error)
// PutBytes is convenience function to put a byte slice into this object
// store under the given name.
//
// ObjectInfo will be returned, containing the object's metadata, digest
// and instance information.
PutBytes(ctx context.Context, name string, data []byte) (*ObjectInfo, error)
// PutString is convenience function to put a string into this object
// store under the given name.
//
// ObjectInfo will be returned, containing the object's metadata, digest
// and instance information.
PutString(ctx context.Context, name string, data string) (*ObjectInfo, error)
// PutFile is convenience function to put a file contents into this
// object store. The name of the object will be the path of the file.
//
// ObjectInfo will be returned, containing the object's metadata, digest
// and instance information.
PutFile(ctx context.Context, file string) (*ObjectInfo, error)
// Get will pull the named object from the object store. If the object
// does not exist, ErrObjectNotFound will be returned.
//
// The returned ObjectResult will contain the object's metadata and a
// reader to read the object's contents. The reader will be closed when
// all data has been read or an error occurs.
//
// A GetObjectShowDeleted option can be supplied to return an object
// even if it was marked as deleted.
Get(ctx context.Context, name string, opts ...GetObjectOpt) (ObjectResult, error)
// GetBytes is a convenience function to pull an object from this object
// store and return it as a byte slice.
//
// If the object does not exist, ErrObjectNotFound will be returned.
//
// A GetObjectShowDeleted option can be supplied to return an object
// even if it was marked as deleted.
GetBytes(ctx context.Context, name string, opts ...GetObjectOpt) ([]byte, error)
// GetString is a convenience function to pull an object from this
// object store and return it as a string.
//
// If the object does not exist, ErrObjectNotFound will be returned.
//
// A GetObjectShowDeleted option can be supplied to return an object
// even if it was marked as deleted.
GetString(ctx context.Context, name string, opts ...GetObjectOpt) (string, error)
// GetFile is a convenience function to pull an object from this object
// store and place it in a file. If the file already exists, it will be
// overwritten, otherwise it will be created.
//
// If the object does not exist, ErrObjectNotFound will be returned.
// A GetObjectShowDeleted option can be supplied to return an object
// even if it was marked as deleted.
GetFile(ctx context.Context, name, file string, opts ...GetObjectOpt) error
// GetInfo will retrieve the current information for the object, containing
// the object's metadata and instance information.
//
// If the object does not exist, ErrObjectNotFound will be returned.
//
// A GetObjectInfoShowDeleted option can be supplied to return an object
// even if it was marked as deleted.
GetInfo(ctx context.Context, name string, opts ...GetObjectInfoOpt) (*ObjectInfo, error)
// UpdateMeta will update the metadata for the object.
//
// If the object does not exist, ErrUpdateMetaDeleted will be returned.
// If the new name is different from the old name, and an object with the
// new name already exists, ErrObjectAlreadyExists will be returned.
UpdateMeta(ctx context.Context, name string, meta ObjectMeta) error
// Delete will delete the named object from the object store. If the object
// does not exist, ErrObjectNotFound will be returned. If the object is
// already deleted, no error will be returned.
//
// All chunks for the object will be purged, and the object will be marked
// as deleted.
Delete(ctx context.Context, name string) error
// AddLink will add a link to another object. A link is a reference to
// another object. The provided name is the name of the link object.
// The provided ObjectInfo is the info of the object being linked to.
//
// If an object with given name already exists, ErrObjectAlreadyExists
// will be returned.
// If object being linked to is deleted, ErrNoLinkToDeleted will be
// returned.
// If the provided object is a link, ErrNoLinkToLink will be returned.
// If the provided object is nil or the name is empty, ErrObjectRequired
// will be returned.
AddLink(ctx context.Context, name string, obj *ObjectInfo) (*ObjectInfo, error)
// AddBucketLink will add a link to another object store. A link is a
// reference to another object store. The provided name is the name of
// the link object.
// The provided ObjectStore is the object store being linked to.
//
// If an object with given name already exists, ErrObjectAlreadyExists
// will be returned.
// If the provided object store is nil ErrBucketRequired will be returned.
AddBucketLink(ctx context.Context, name string, bucket ObjectStore) (*ObjectInfo, error)
// Seal will seal the object store, no further modifications will be allowed.
Seal(ctx context.Context) error
// Watch for any updates to objects in the store. By default, the watcher will send the latest
// info for each object and all future updates. Watch will send a nil
// entry when it has received all initial values. There are a few ways
// to configure the watcher:
//
// - IncludeHistory will have the watcher send all historical information
// for each object.
// - IgnoreDeletes will have the watcher not pass any objects with
// delete markers.
// - UpdatesOnly will have the watcher only pass updates on objects
// (without latest info when started).
Watch(ctx context.Context, opts ...WatchOpt) (ObjectWatcher, error)
// List will list information about objects in the store.
//
// If the object store is empty, ErrNoObjectsFound will be returned.
List(ctx context.Context, opts ...ListObjectsOpt) ([]*ObjectInfo, error)
// Status retrieves the status and configuration of the bucket.
Status(ctx context.Context) (ObjectStoreStatus, error)
}
// ObjectWatcher is what is returned when doing a watch. It can be used to
// retrieve updates to objects in a bucket. If not using UpdatesOnly option,
// it will also send the latest value for each key. After all initial values
// have been sent, a nil entry will be sent. Stop can be used to stop the
// watcher and close the underlying channel. Watcher will not close the
// channel until Stop is called or connection is closed.
ObjectWatcher interface {
Updates() <-chan *ObjectInfo
Stop() error
}
// ObjectStoreConfig is the configuration for the object store.
ObjectStoreConfig struct {
// Bucket is the name of the object store. Bucket name has to be
// unique and can only contain alphanumeric characters, dashes, and
// underscores.
Bucket string `json:"bucket"`
// Description is an optional description for the object store.
Description string `json:"description,omitempty"`
// TTL is the maximum age of objects in the store. If an object is not
// updated within this time, it will be removed from the store.
// By default, objects do not expire.
TTL time.Duration `json:"max_age,omitempty"`
// MaxBytes is the maximum size of the object store. If not specified,
// the default is -1 (unlimited).
MaxBytes int64 `json:"max_bytes,omitempty"`
// Storage is the type of storage to use for the object store. If not
// specified, the default is FileStorage.
Storage StorageType `json:"storage,omitempty"`
// Replicas is the number of replicas to keep for the object store in
// clustered jetstream. Defaults to 1, maximum is 5.
Replicas int `json:"num_replicas,omitempty"`
// Placement is used to declare where the object store should be placed via
// tags and/or an explicit cluster name.
Placement *Placement `json:"placement,omitempty"`
// Compression enables the underlying stream compression.
// NOTE: Compression is supported for nats-server 2.10.0+
Compression bool `json:"compression,omitempty"`
// Bucket-specific metadata
// NOTE: Metadata requires nats-server v2.10.0+
Metadata map[string]string `json:"metadata,omitempty"`
}
// ObjectStoresLister is used to retrieve a list of object stores. It returns
// a channel to read the bucket store statuses from. The lister will always
// close the channel when done (either all stores have been retrieved or an
// error occurred) and therefore can be used in range loops. Stop can be
// used to stop the lister when not all object stores have been read.
ObjectStoresLister interface {
Status() <-chan ObjectStoreStatus
Error() error
}
// ObjectStoreNamesLister is used to retrieve a list of object store names.
// It returns a channel to read the bucket names from. The lister will
// always close the channel when done (either all stores have been retrieved
// or an error occurred) and therefore can be used in range loops. Stop can
// be used to stop the lister when not all bucket names have been read.
ObjectStoreNamesLister interface {
Name() <-chan string
Error() error
}
// ObjectStoreStatus is run-time status about a bucket.
ObjectStoreStatus interface {
// Bucket returns the name of the object store.
Bucket() string
// Description is the description supplied when creating the bucket.
Description() string
// TTL indicates how long objects are kept in the bucket.
TTL() time.Duration
// Storage indicates the underlying JetStream storage technology used to
// store data.
Storage() StorageType
// Replicas indicates how many storage replicas are kept for the data in
// the bucket.
Replicas() int
// Sealed indicates the stream is sealed and cannot be modified in any
// way.
Sealed() bool
// Size is the combined size of all data in the bucket including
// metadata, in bytes.
Size() uint64
// BackingStore indicates what technology is used for storage of the
// bucket. Currently only JetStream is supported.
BackingStore() string
// Metadata is the user supplied metadata for the bucket.
Metadata() map[string]string
// IsCompressed indicates if the data is compressed on disk.
IsCompressed() bool
}
// ObjectMetaOptions is used to set additional options when creating an object.
ObjectMetaOptions struct {
// Link contains information about a link to another object or object store.
// It should not be set manually, but rather by using the AddLink or
// AddBucketLink methods.
Link *ObjectLink `json:"link,omitempty"`
// ChunkSize is the maximum size of each chunk in bytes. If not specified,
// the default is 128k.
ChunkSize uint32 `json:"max_chunk_size,omitempty"`
}
// ObjectMeta is high level information about an object.
ObjectMeta struct {
// Name is the name of the object. The name is required when adding an
// object and has to be unique within the object store.
Name string `json:"name"`
// Description is an optional description for the object.
Description string `json:"description,omitempty"`
// Headers is an optional set of user-defined headers for the object.
Headers nats.Header `json:"headers,omitempty"`
// Metadata is the user supplied metadata for the object.
Metadata map[string]string `json:"metadata,omitempty"`
// Additional options for the object.
Opts *ObjectMetaOptions `json:"options,omitempty"`
}
// ObjectInfo contains ObjectMeta and additional information about an
// object.
ObjectInfo struct {
// ObjectMeta contains high level information about the object.
ObjectMeta
// Bucket is the name of the object store.
Bucket string `json:"bucket"`
// NUID is the unique identifier for the object set when putting the
// object into the store.
NUID string `json:"nuid"`
// Size is the size of the object in bytes. It only includes the size of
// the object itself, not the metadata.
Size uint64 `json:"size"`
// ModTime is the last modification time of the object.
ModTime time.Time `json:"mtime"`
// Chunks is the number of chunks the object is split into. Maximum size
// of each chunk can be specified in ObjectMetaOptions.
Chunks uint32 `json:"chunks"`
// Digest is the SHA-256 digest of the object. It is used to verify the
// integrity of the object.
Digest string `json:"digest,omitempty"`
// Deleted indicates if the object is marked as deleted.
Deleted bool `json:"deleted,omitempty"`
}
// ObjectLink is used to embed links to other buckets and objects.
ObjectLink struct {
// Bucket is the name of the object store the link is pointing to.
Bucket string `json:"bucket"`
// Name can be used to link to a single object.
// If empty means this is a link to the whole store, like a directory.
Name string `json:"name,omitempty"`
}
// ObjectResult will return the object info and a reader to read the object's
// contents. The reader will be closed when all data has been read or an
// error occurs.
ObjectResult interface {
io.ReadCloser
Info() (*ObjectInfo, error)
Error() error
}
// GetObjectOpt is used to set additional options when getting an object.
GetObjectOpt func(opts *getObjectOpts) error
// GetObjectInfoOpt is used to set additional options when getting object info.
GetObjectInfoOpt func(opts *getObjectInfoOpts) error
// ListObjectsOpt is used to set additional options when listing objects.
ListObjectsOpt func(opts *listObjectOpts) error
getObjectOpts struct {
// Include deleted object in the result.
showDeleted bool
}
getObjectInfoOpts struct {
// Include deleted object in the result.
showDeleted bool
}
listObjectOpts struct {
// Include deleted objects in the result channel.
showDeleted bool
}
obs struct {
name string
streamName string
stream Stream
pushJS nats.JetStreamContext
js *jetStream
}
// ObjectResult impl.
objResult struct {
sync.Mutex
info *ObjectInfo
r io.ReadCloser
err error
ctx context.Context
digest hash.Hash
}
)
const (
objNameTmpl = "OBJ_%s" // OBJ_<bucket> // stream name
objAllChunksPreTmpl = "$O.%s.C.>" // $O.<bucket>.C.> // chunk stream subject
objAllMetaPreTmpl = "$O.%s.M.>" // $O.<bucket>.M.> // meta stream subject
objChunksPreTmpl = "$O.%s.C.%s" // $O.<bucket>.C.<object-nuid> // chunk message subject
objMetaPreTmpl = "$O.%s.M.%s" // $O.<bucket>.M.<name-encoded> // meta message subject
objNoPending = "0"
objDefaultChunkSize = uint32(128 * 1024) // 128k
objDigestType = "SHA-256="
objDigestTmpl = objDigestType + "%s"
)
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 StreamConfig{}, ErrInvalidStoreName
}
name := cfg.Bucket
chunks := fmt.Sprintf(objAllChunksPreTmpl, name)
meta := fmt.Sprintf(objAllMetaPreTmpl, name)
// We will set explicitly some values so that we can do comparison
// if we get an "already in use" error and need to check if it is same.
// See kv
replicas := cfg.Replicas
if replicas == 0 {
replicas = 1
}
maxBytes := cfg.MaxBytes
if maxBytes == 0 {
maxBytes = -1
}
var compression StoreCompression
if cfg.Compression {
compression = S2Compression
}
scfg := StreamConfig{
Name: fmt.Sprintf(objNameTmpl, name),
Description: cfg.Description,
Subjects: []string{chunks, meta},
MaxAge: cfg.TTL,
MaxBytes: maxBytes,
Storage: cfg.Storage,
Replicas: replicas,
Placement: cfg.Placement,
Discard: DiscardNew,
AllowRollup: true,
AllowDirect: true,
Metadata: cfg.Metadata,
Compression: compression,
}
return scfg, nil
}
// ObjectStore will look up and bind to an existing object store instance.
func (js *jetStream) ObjectStore(ctx context.Context, bucket string) (ObjectStore, error) {
if !validBucketRe.MatchString(bucket) {
return nil, ErrInvalidStoreName
}
streamName := fmt.Sprintf(objNameTmpl, bucket)
stream, err := js.Stream(ctx, streamName)
if err != nil {
if errors.Is(err, ErrStreamNotFound) {
err = ErrBucketNotFound
}
return nil, err
}
pushJS, err := js.legacyJetStream()
if err != nil {
return nil, err
}
return mapStreamToObjectStore(js, pushJS, bucket, stream), nil
}
// DeleteObjectStore will delete the underlying stream for the named object.
func (js *jetStream) DeleteObjectStore(ctx context.Context, bucket string) error {
stream := fmt.Sprintf(objNameTmpl, bucket)
return js.DeleteStream(ctx, stream)
}
func encodeName(name string) string {
return base64.URLEncoding.EncodeToString([]byte(name))
}
// Put will place the contents from the reader into this object-store.
func (obs *obs) Put(ctx context.Context, meta ObjectMeta, r io.Reader) (*ObjectInfo, error) {
if meta.Name == "" {
return nil, ErrBadObjectMeta
}
if meta.Opts == nil {
meta.Opts = &ObjectMetaOptions{ChunkSize: objDefaultChunkSize}
} else if meta.Opts.Link != nil {
return nil, ErrLinkNotAllowed
} else if meta.Opts.ChunkSize == 0 {
meta.Opts.ChunkSize = objDefaultChunkSize
}
// Create the new nuid so chunks go on a new subject if the name is re-used
newnuid := nuid.Next()
// These will be used in more than one place
chunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, newnuid)
// Grab existing meta info (einfo). Ok to be found or not found, any other error is a problem
// Chunks on the old nuid can be cleaned up at the end
einfo, err := obs.GetInfo(ctx, meta.Name, GetObjectInfoShowDeleted()) // GetInfo will encode the name
if err != nil && err != ErrObjectNotFound {
return nil, err
}
// For async error handling
var perr error
var mu sync.Mutex
setErr := func(err error) {
mu.Lock()
defer mu.Unlock()
perr = err
}
getErr := func() error {
mu.Lock()
defer mu.Unlock()
return perr
}
// Create our own JS context to handle errors etc.
pubJS, err := New(obs.js.conn, WithPublishAsyncErrHandler(func(js JetStream, _ *nats.Msg, err error) { setErr(err) }))
if err != nil {
return nil, err
}
defer pubJS.(*jetStream).cleanupReplySub()
purgePartial := func() {
// wait until all pubs are complete or up to default timeout before attempting purge
select {
case <-pubJS.PublishAsyncComplete():
case <-ctx.Done():
}
_ = obs.stream.Purge(ctx, WithPurgeSubject(chunkSubj))
}
m, h := nats.NewMsg(chunkSubj), sha256.New()
chunk, sent, total := make([]byte, meta.Opts.ChunkSize), 0, uint64(0)
// set up the info object. The chunk upload sets the size and digest
info := &ObjectInfo{Bucket: obs.name, NUID: newnuid, ObjectMeta: meta}
for r != nil {
if ctx != nil {
select {
case <-ctx.Done():
if ctx.Err() == context.Canceled {
err = ctx.Err()
} else {
err = nats.ErrTimeout
}
default:
}
if err != nil {
purgePartial()
return nil, err
}
}
// Actual read.
// TODO(dlc) - Deadline?
n, readErr := r.Read(chunk)
// Handle all non EOF errors
if readErr != nil && readErr != io.EOF {
purgePartial()
return nil, readErr
}
// Add chunk only if we received data
if n > 0 {
// Chunk processing.
m.Data = chunk[:n]
h.Write(m.Data)
// Send msg itself.
if _, err := pubJS.PublishMsgAsync(m); err != nil {
purgePartial()
return nil, err
}
if err := getErr(); err != nil {
purgePartial()
return nil, err
}
// Update totals.
sent++
total += uint64(n)
}
// EOF Processing.
if readErr == io.EOF {
// Place meta info.
info.Size, info.Chunks = uint64(total), uint32(sent)
info.Digest = GetObjectDigestValue(h)
break
}
}
// Prepare the meta message
metaSubj := fmt.Sprintf(objMetaPreTmpl, obs.name, encodeName(meta.Name))
mm := nats.NewMsg(metaSubj)
mm.Header.Set(MsgRollup, MsgRollupSubject)
mm.Data, err = json.Marshal(info)
if err != nil {
if r != nil {
purgePartial()
}
return nil, err
}
// Publish the meta message.
_, err = pubJS.PublishMsgAsync(mm)
if err != nil {
if r != nil {
purgePartial()
}
return nil, err
}
// Wait for all to be processed.
select {
case <-pubJS.PublishAsyncComplete():
if err := getErr(); err != nil {
if r != nil {
purgePartial()
}
return nil, err
}
case <-ctx.Done():
return nil, nats.ErrTimeout
}
info.ModTime = time.Now().UTC() // This time is not actually the correct time
// Delete any original chunks.
if einfo != nil && !einfo.Deleted {
echunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, einfo.NUID)
_ = obs.stream.Purge(ctx, WithPurgeSubject(echunkSubj))
}
// TODO would it be okay to do this to return the info with the correct time?
// With the understanding that it is an extra call to the server.
// Otherwise the time the user gets back is the client time, not the server time.
// return obs.GetInfo(info.Name)
return info, nil
}
// GetObjectDigestValue calculates the base64 value of hashed data
func GetObjectDigestValue(data hash.Hash) string {
sha := data.Sum(nil)
return fmt.Sprintf(objDigestTmpl, base64.URLEncoding.EncodeToString(sha[:]))
}
// DecodeObjectDigest decodes base64 hash
func DecodeObjectDigest(data string) ([]byte, error) {
digest := strings.SplitN(data, "=", 2)
if len(digest) != 2 {
return nil, ErrInvalidDigestFormat
}
return base64.URLEncoding.DecodeString(digest[1])
}
func (info *ObjectInfo) isLink() bool {
return info.ObjectMeta.Opts != nil && info.ObjectMeta.Opts.Link != nil
}
// Get will pull the object from the underlying stream.
func (obs *obs) Get(ctx context.Context, name string, opts ...GetObjectOpt) (ObjectResult, error) {
var o getObjectOpts
for _, opt := range opts {
if opt != nil {
if err := opt(&o); err != nil {
return nil, err
}
}
}
infoOpts := make([]GetObjectInfoOpt, 0)
if o.showDeleted {
infoOpts = append(infoOpts, GetObjectInfoShowDeleted())
}
// Grab meta info.
info, err := obs.GetInfo(ctx, name, infoOpts...)
if err != nil {
return nil, err
}
if info.NUID == "" {
return nil, ErrBadObjectMeta
}
// Check for object links. If single objects we do a pass through.
if info.isLink() {
if info.ObjectMeta.Opts.Link.Name == "" {
return nil, ErrCantGetBucket
}
// is the link in the same bucket?
lbuck := info.ObjectMeta.Opts.Link.Bucket
if lbuck == obs.name {
return obs.Get(ctx, info.ObjectMeta.Opts.Link.Name)
}
// different bucket
lobs, err := obs.js.ObjectStore(ctx, lbuck)
if err != nil {
return nil, err
}
return lobs.Get(ctx, info.ObjectMeta.Opts.Link.Name)
}
result := &objResult{info: info, ctx: ctx}
if info.Size == 0 {
return result, nil
}
pr, pw := net.Pipe()
result.r = pr
gotErr := func(m *nats.Msg, err error) {
pw.Close()
m.Sub.Unsubscribe()
result.setErr(err)
}
// For calculating sum256
result.digest = sha256.New()
processChunk := func(m *nats.Msg) {
var err error
if ctx != nil {
select {
case <-ctx.Done():
if ctx.Err() == context.Canceled {
err = ctx.Err()
} else {
err = nats.ErrTimeout
}
default:
}
if err != nil {
gotErr(m, err)
return
}
}
tokens, err := parser.GetMetadataFields(m.Reply)
if err != nil {
gotErr(m, err)
return
}
// Write to our pipe.
for b := m.Data; len(b) > 0; {
n, err := pw.Write(b)
if err != nil {
gotErr(m, err)
return
}
b = b[n:]
}
// Update sha256
result.digest.Write(m.Data)
// Check if we are done.
if tokens[parser.AckNumPendingTokenPos] == objNoPending {
pw.Close()
m.Sub.Unsubscribe()
}
}
chunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, info.NUID)
streamName := fmt.Sprintf(objNameTmpl, obs.name)
subscribeOpts := []nats.SubOpt{
nats.OrderedConsumer(),
nats.Context(ctx),
nats.BindStream(streamName),
}
_, err = obs.pushJS.Subscribe(chunkSubj, processChunk, subscribeOpts...)
if err != nil {
return nil, err
}
return result, nil
}
// Delete will delete the object.
func (obs *obs) Delete(ctx context.Context, name string) error {
// Grab meta info.
info, err := obs.GetInfo(ctx, name, GetObjectInfoShowDeleted())
if err != nil {
return err
}
if info.NUID == "" {
return ErrBadObjectMeta
}
// Place a rollup delete marker and publish the info
info.Deleted = true
info.Size, info.Chunks, info.Digest = 0, 0, ""
if err = publishMeta(ctx, info, obs.js); err != nil {
return err
}
// Purge chunks for the object.
chunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, info.NUID)
return obs.stream.Purge(ctx, WithPurgeSubject(chunkSubj))
}
func publishMeta(ctx context.Context, info *ObjectInfo, js *jetStream) error {
// marshal the object into json, don't store an actual time
info.ModTime = time.Time{}
data, err := json.Marshal(info)
if err != nil {
return err
}
// Prepare and publish the message.
mm := nats.NewMsg(fmt.Sprintf(objMetaPreTmpl, info.Bucket, encodeName(info.ObjectMeta.Name)))
mm.Header.Set(MsgRollup, MsgRollupSubject)
mm.Data = data
if _, err := js.PublishMsg(ctx, mm); err != nil {
return err
}
// set the ModTime in case it's returned to the user, even though it's not the correct time.
info.ModTime = time.Now().UTC()
return nil
}
// AddLink will add a link to another object if it's not deleted and not another link
// name is the name of this link object
// obj is what is being linked too
func (obs *obs) AddLink(ctx context.Context, name string, obj *ObjectInfo) (*ObjectInfo, error) {
if name == "" {
return nil, ErrNameRequired
}
// TODO Handle stale info
if obj == nil || obj.Name == "" {
return nil, ErrObjectRequired
}
if obj.Deleted {
return nil, ErrNoLinkToDeleted
}
if obj.isLink() {
return nil, ErrNoLinkToLink
}
// If object with link's name is found, error.