func (u *Uploader) UploadBytes(ctx context.Context, r io.Reader) (blob.Ref, error) {
	bb := schema.NewBuilder()
	bb.SetType("bytes")
	err := schema.WriteFileChunks(ctx, u.StatReceiver, bb, r)
	if err != nil {
		return blob.Ref{}, err
	}
	b := bb.Blob()
	br, err := u.Client.UploadBlob(ctx, b)
	return br.BlobRef, err
}
UploadBytes uploads the contents of the io.Reader as "bytes" blob.
package camutil
import (
	"bytes"
	"context"
	"encoding/base64"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"strings"
	"sync"
	"time"
	"golang.org/x/crypto/openpgp"
	"golang.org/x/crypto/openpgp/armor"
	"perkeep.org/pkg/blob"
	"perkeep.org/pkg/blobserver"
	"perkeep.org/pkg/blobserver/localdisk"
	"perkeep.org/pkg/client"
	"perkeep.org/pkg/schema"
)
type Uploader struct {
	*client.Client
	blobserver.StatReceiver
	*schema.Signer
	gate          chan struct{}
	server        string
	args          []string
	opts          []string
	env           []string
	mtx           sync.Mutex
	skipHaveCache bool
}
var ErrFileIsEmpty = errors.New("file is empty")
var cachedUploader = make(map[string]*Uploader, 1)
var cachedUploaderMtx = new(sync.Mutex)
func Close() error {
	cachedUploaderMtx.Lock()
	defer cachedUploaderMtx.Unlock()
	for k := range cachedUploader {
		delete(cachedUploader, k)
	}
	cachedDownloaderMtx.Lock()
	defer cachedDownloaderMtx.Unlock()
	for k := range cachedDownloader {
		cachedDownloader[k].Close()
		delete(cachedDownloader, k)
	}
	return nil
}
func NewUploader(server string, capCtime bool, skipHaveCache bool) *Uploader {
	key := fmt.Sprintf("%q\t%t\t%t", server, capCtime, skipHaveCache)
	cachedUploaderMtx.Lock()
	defer cachedUploaderMtx.Unlock()
	u, ok := cachedUploader[key]
	if ok {
		return u
	}
	maxProcs := runtime.GOMAXPROCS(-1)
	if strings.HasPrefix(server, "file://") {
		recv, err := localdisk.New(server[7:])
		if err != nil {
			Log("msg", "localdisk.New", "server", server, "error", err)
			return nil
		}
		u = &Uploader{
			server:        server,
			gate:          make(chan struct{}, maxProcs),
			skipHaveCache: skipHaveCache,
			StatReceiver:  recv,
			Signer:        newDummySigner(),
		}
		cachedUploader[key] = u
		return u
	}
	c, err := NewClient(server)
	if err != nil || c == nil {
		Log("msg", "NewClient", "server", server, "error", err)
		return nil
	}
	u = &Uploader{
		server:        server,
		args:          make([]string, 1, 2),
		opts:          make([]string, 0, 3),
		gate:          make(chan struct{}, maxProcs),
		skipHaveCache: skipHaveCache,
		Client:        c,
		StatReceiver:  c,
	}
	u.args[0] = cmdPkPut
	if server != "" {
		u.args = append(u.args, "-server="+server)
	}
	needDebugEnv := false
	if skipHaveCache {
		u.opts = append(u.opts, "-havecache=false", "-statcache=false")
		needDebugEnv = true
	}
	if capCtime {
		u.opts = append(u.opts, "-capctime")
		needDebugEnv = true
	}
	if needDebugEnv {
		if os.Getenv("CAMLI_DEBUG") != "true" {  
			u.env = append(os.Environ(), "CAMLI_DEBUG=true")
		}
	}
	cachedUploader[key] = u
	return u
}
func (u *Uploader) Close() error {
	var err error
	if u.StatReceiver != nil {
		if cl, ok := u.StatReceiver.(io.Closer); ok {
			err = cl.Close()
		}
		u.StatReceiver = nil
	}
	if u.Client == nil {
		return err
	}
	defer func() {
		if r := recover(); r != nil {
			err = r.(error)
		}
	}()
	err = u.Client.Close()
	return err
}