Golang camlistore-org-pkg-blobserver.GetHub类(方法)实例源码

func TestReceive(t *testing.T) {
	sto := new(test.Fetcher)
	data := []byte("some blob")
	br := blob.SHA1FromBytes(data)

	hub := blobserver.GetHub(sto)
	ch := make(chan blob.Ref, 1)
	hub.RegisterListener(ch)

	sb, err := blobserver.Receive(sto, br, bytes.NewReader(data))
	if err != nil {
		t.Fatal(err)
	}
	if sb.Size != int64(len(data)) {
		t.Errorf("received blob size = %d; want %d", sb.Size, len(data))
	}
	if sb.Ref != br {
		t.Errorf("received blob = %v; want %v", sb.Ref, br)
	}
	select {
	case got := <-ch:
		if got != br {
			t.Errorf("blobhub notified about %v; want %v", got, br)
		}
	case <-time.After(5 * time.Second):
		t.Error("timeout waiting on blobhub")
	}
}

// NewSyncHandler returns a handler that will asynchronously and continuously
// copy blobs from src to dest, if missing on dest.
// Blobs waiting to be copied are stored on pendingQueue. srcName and destName are
// only used for status and debugging messages.
// N.B: blobs should be added to src with a method that notifies the blob hub,
// such as blobserver.Receive.
func NewSyncHandler(srcName, destName string,
	src blobserver.Storage, dest blobReceiverEnumerator,
	pendingQueue sorted.KeyValue) *SyncHandler {
	sh := newSyncHandler(srcName, destName, src, dest, pendingQueue)
	go sh.syncLoop()
	blobserver.GetHub(sh.from).AddReceiveHook(sh.enqueue)
	return sh
}

func (sh *Handler) subscribeToNewBlobs() {
	ch := make(chan blob.Ref, buffered)
	blobserver.GetHub(sh.index).RegisterListener(ch)
	go func() {
		for br := range ch {
			bm, err := sh.index.GetBlobMeta(br)
			if err == nil {
				sh.wsHub.newBlobRecv <- bm.CamliType
			}
		}
	}()
}

func (sh *Handler) subscribeToNewBlobs() {
	ch := make(chan blob.Ref, buffered)
	blobserver.GetHub(sh.index).RegisterListener(ch)
	go func() {
		ctx := context.Background()
		for br := range ch {
			sh.index.RLock()
			bm, err := sh.index.GetBlobMeta(ctx, br)
			if err == nil {
				sh.wsHub.newBlobRecv <- bm.CamliType
			}
			sh.index.RUnlock()
		}
	}()
}

func newSyncFromConfig(ld blobserver.Loader, conf jsonconfig.Obj) (http.Handler, error) {
	var (
		from           = conf.RequiredString("from")
		to             = conf.RequiredString("to")
		fullSync       = conf.OptionalBool("fullSyncOnStart", false)
		blockFullSync  = conf.OptionalBool("blockingFullSyncOnStart", false)
		idle           = conf.OptionalBool("idle", false)
		queueConf      = conf.OptionalObject("queue")
		copierPoolSize = conf.OptionalInt("copierPoolSize", 5)
		validate       = conf.OptionalBool("validateOnStart", validateOnStartDefault)
	)
	if err := conf.Validate(); err != nil {
		return nil, err
	}
	if idle {
		return newIdleSyncHandler(from, to), nil
	}
	if len(queueConf) == 0 {
		return nil, errors.New(`Missing required "queue" object`)
	}
	q, err := sorted.NewKeyValue(queueConf)
	if err != nil {
		return nil, err
	}

	isToIndex := false
	fromBs, err := ld.GetStorage(from)
	if err != nil {
		return nil, err
	}
	toBs, err := ld.GetStorage(to)
	if err != nil {
		return nil, err
	}
	if _, ok := fromBs.(*index.Index); !ok {
		if _, ok := toBs.(*index.Index); ok {
			isToIndex = true
		}
	}

	sh := newSyncHandler(from, to, fromBs, toBs, q)
	sh.toIndex = isToIndex
	sh.copierPoolSize = copierPoolSize
	if err := sh.readQueueToMemory(); err != nil {
		return nil, fmt.Errorf("Error reading sync queue to memory: %v", err)
	}

	if fullSync || blockFullSync {
		sh.logf("Doing full sync")
		didFullSync := make(chan bool, 1)
		go func() {
			for {
				n := sh.runSync("queue", sh.enumeratePendingBlobs)
				if n > 0 {
					sh.logf("Queue sync copied %d blobs", n)
					continue
				}
				break
			}
			n := sh.runSync("full", blobserverEnumerator(context.TODO(), fromBs))
			sh.logf("Full sync copied %d blobs", n)
			didFullSync <- true
			sh.syncLoop()
		}()
		if blockFullSync {
			sh.logf("Blocking startup, waiting for full sync from %q to %q", from, to)
			<-didFullSync
			sh.logf("Full sync complete.")
		}
	} else {
		go sh.syncLoop()
	}

	if validate {
		go sh.startFullValidation()
	}

	blobserver.GetHub(fromBs).AddReceiveHook(sh.enqueue)
	return sh, nil
}

func newSyncFromConfig(ld blobserver.Loader, conf jsonconfig.Obj) (http.Handler, error) {
	var (
		from          = conf.RequiredString("from")
		to            = conf.RequiredString("to")
		fullSync      = conf.OptionalBool("fullSyncOnStart", false)
		blockFullSync = conf.OptionalBool("blockingFullSyncOnStart", false)
		idle          = conf.OptionalBool("idle", false)
		queueConf     = conf.OptionalObject("queue")
	)
	if err := conf.Validate(); err != nil {
		return nil, err
	}
	if idle {
		synch, err := createIdleSyncHandler(from, to)
		if err != nil {
			return nil, err
		}
		return synch, nil
	}
	if len(queueConf) == 0 {
		return nil, errors.New(`Missing required "queue" object`)
	}
	q, err := sorted.NewKeyValue(queueConf)
	if err != nil {
		return nil, err
	}

	isToIndex := false
	fromBs, err := ld.GetStorage(from)
	if err != nil {
		return nil, err
	}
	toBs, err := ld.GetStorage(to)
	if err != nil {
		return nil, err
	}
	if _, ok := fromBs.(*index.Index); !ok {
		if _, ok := toBs.(*index.Index); ok {
			isToIndex = true
		}
	}

	sh, err := createSyncHandler(from, to, fromBs, toBs, q, isToIndex)
	if err != nil {
		return nil, err
	}

	if fullSync || blockFullSync {
		didFullSync := make(chan bool, 1)
		go func() {
			n := sh.runSync("queue", sh.enumerateQueuedBlobs)
			sh.logf("Queue sync copied %d blobs", n)
			n = sh.runSync("full", blobserverEnumerator(context.TODO(), fromBs))
			sh.logf("Full sync copied %d blobs", n)
			didFullSync <- true
			sh.syncQueueLoop()
		}()
		if blockFullSync {
			sh.logf("Blocking startup, waiting for full sync from %q to %q", from, to)
			<-didFullSync
			sh.logf("Full sync complete.")
		}
	} else {
		go sh.syncQueueLoop()
	}

	blobserver.GetHub(fromBs).AddReceiveHook(sh.enqueue)
	return sh, nil
}