Method Put

object.go:331–527

Put will place the contents from the reader into this object-store.

(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt)

// Put will place the contents from the reader into this object-store.
func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectInfo, error) {
	if meta == nil || 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
	}

	var o objOpts
	for _, opt := range opts {
		if opt != nil {
			if err := opt.configureObject(&o); err != nil {
				return nil, err
			}
		}
	}
	ctx := o.ctx

	// 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(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.
	jetStream, err := obs.js.nc.JetStream(PublishAsyncErrHandler(func(js JetStream, _ *Msg, err error) { setErr(err) }))
	if err != nil {
		return nil, err
	}

	defer jetStream.(*js).cleanupReplySub()