// LastPullSubscriptionChunk returns ChunkDescriptor of the latest Chunk
// in pull syncing index for a provided bin. If there are no chunks in
// that bin, chunk.ErrChunkNotFound is returned.
975293
func (db *DB) LastPullSubscriptionChunk(bin uint8) (c *ChunkDescriptor, err error) {
	item, err := db.pullIndex.Last([]byte{bin})
	if err != nil {
		if err == leveldb.ErrNotFound {
			return nil, chunk.ErrChunkNotFound
		}
		return nil, err
	}
	return &ChunkDescriptor{
		Address:        item.Address,
		StoreTimestamp: item.StoreTimestamp,
	}, nil
}
1285879
func (r *marathonClient) HasSubscription(callback string) (bool, error) {
	// step: generate our events callback
	subscriptions, err := r.Subscriptions()
	if err != nil {
		return false, err
	}

	for _, subscription := range subscriptions.CallbackURLs {
		if callback == subscription {
			return true, nil
		}
	}

	return false, nil
}
498178
func (s *StreamingReader) signChunk(chunkLen int) {
	// Compute chunk signature for next header
	signature := buildChunkSignature(s.chunkBuf[:chunkLen], s.reqTime,
		s.region, s.prevSignature, s.secretAccessKey)

	// For next chunk signature computation
	s.prevSignature = signature

	// Write chunk header into streaming buffer
	chunkHdr := buildChunkHeader(int64(chunkLen), signature)
	s.buf.Write(chunkHdr)

	// Write chunk data into streaming buffer
	s.buf.Write(s.chunkBuf[:chunkLen])

	// Write the chunk trailer.
	s.buf.Write([]byte("\r\n"))

	// Reset chunkBufLen for next chunk read.
	s.chunkBufLen = 0
	s.chunkNum++
}
1373938
func (ef *File) UnorderedChunks() (cc chan Chunk) {
	cc = make(chan Chunk)
	go func() {
		defer close(cc)
		for i := uint16(0); i < ef.Header.ChunkCount; i++ {
			offsetChunk := int64(ef.Header.ChunkDataOffset) + int64(ChunkSize)*int64(i)
			//chunk, err := ef.FetchChunk(offsetChunk)
			chunk, err := ef.FetchRawChunk(offsetChunk)
			switch {
			case err != nil && err != io.EOF:
				panic(err)
			case err == nil:
				cc <- chunk
			}
		}
	}()
	return
}
1281534
func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *ChunkDeliveryMsg) error {
	var osp opentracing.Span
	ctx, osp = spancontext.StartSpan(
		ctx,
		"handle.chunk.delivery")

	processReceivedChunksCount.Inc(1)

	// retrieve the span for the originating retrieverequest
	spanId := fmt.Sprintf("stream.send.request.%v.%v", sp.ID(), req.Addr)
	span := tracing.ShiftSpanByKey(spanId)

	log.Trace("handle.chunk.delivery", "ref", req.Addr, "from peer", sp.ID())

	go func() {
		defer osp.Finish()

		if span != nil {
			span.LogFields(olog.String("finish", "from handleChunkDeliveryMsg"))
			defer span.Finish()
		}

		req.peer = sp
		log.Trace("handle.chunk.delivery", "put", req.Addr)
		err := d.chunkStore.Put(ctx, storage.NewChunk(req.Addr, req.SData))
		if err != nil {
			if err == storage.ErrChunkInvalid {
				// we removed this log because it spams the logs
				// TODO: Enable this log line
				// log.Warn("invalid chunk delivered", "peer", sp.ID(), "chunk", req.Addr, )
				req.peer.Drop(err)
			}
		}
		log.Trace("handle.chunk.delivery", "done put", req.Addr, "err", err)
	}()
	return nil
}
972510
func (e *Error) SubscriptionLoss() (result bool) {
	_, result = CodesIndicatingSubscriptionLoss[e.code]
	return
}
401532
func LastCheckpoint(dir string) (string, int, error) {
	files, err := ioutil.ReadDir(dir)
	if err != nil {
		return "", 0, err
	}
	// Traverse list backwards since there may be multiple checkpoints left.
	for i := len(files) - 1; i >= 0; i-- {
		fi := files[i]

		if !strings.HasPrefix(fi.Name(), checkpointPrefix) {
			continue
		}
		if !fi.IsDir() {
			return "", 0, errors.Errorf("checkpoint %s is not a directory", fi.Name())
		}
		idx, err := strconv.Atoi(fi.Name()[len(checkpointPrefix):])
		if err != nil {
			continue
		}
		return filepath.Join(dir, fi.Name()), idx, nil
	}
	return "", 0, ErrNotFound
}
1085660
func (e *baseExecutor) newFirstChunk() *chunk.Chunk {
	return chunk.New(e.retTypes(), e.initCap, e.maxChunkSize)
}
248057
func (m *SubscriptionManager) cancelSubscription(sub *newSubscription) {
	select {
	case m.cancelSubscriptions <- &cancelSubscription{sub.id}:
	case <-m.quit:
	}
}
1196454
func SplitTagChunks(tags core.TagSlice, stopTag *core.Tag, chunkDelimiter *core.Tag) []core.TagSlice {
	chunks := make([]core.TagSlice, 0)

	tagIndex := 0
	for tagIndex < len(tags) {
		if tags[tagIndex].Equals(stopTag) {
			break
		}

		chunk := make([]*core.Tag, 1)
		chunk[0] = tags[tagIndex]
		tagIndex++

		foundStop := false

		for {
			if tags[tagIndex].Equals(chunkDelimiter) {
				chunk = append(chunk, tags[tagIndex])
				tagIndex++
				break
			}
			if tags[tagIndex].Equals(stopTag) {
				foundStop = true
				tagIndex++
				break
			} else {
				chunk = append(chunk, tags[tagIndex])
				tagIndex++
			}
		}
		chunks = append(chunks, chunk)

		if foundStop {
			break
		}
	}

	return chunks
}
810873
func NewPaperDocChangeSubscriptionDetails(EventUuid string, NewSubscriptionLevel string) *PaperDocChangeSubscriptionDetails {
	s := new(PaperDocChangeSubscriptionDetails)
	s.EventUuid = EventUuid
	s.NewSubscriptionLevel = NewSubscriptionLevel
	return s
}
837186
func (is *store) GetLastUpdated(id ID) (time.Time, error) {
	bytes, err := is.fs.GetMetadata(id.Digest(), "lastUpdated")
	if err != nil || len(bytes) == 0 {
		// No lastUpdated time
		return time.Time{}, nil
	}
	return time.Parse(time.RFC3339Nano, string(bytes))
}
146529
func parseSubscriptionName(rawArgs json.RawMessage) (string, error) {
	dec := json.NewDecoder(bytes.NewReader(rawArgs))
	if tok, _ := dec.Token(); tok != json.Delim('[') {
		return "", errors.New("non-array args")
	}
	v, _ := dec.Token()
	method, ok := v.(string)
	if !ok {
		return "", errors.New("expected subscription name as first argument")
	}
	return method, nil
}
691167
func (w *WSEvents) redoSubscriptionsAfter(d time.Duration) {
	time.Sleep(d)

	for q := range w.subscriptions {
		err := w.ws.Subscribe(context.Background(), q)
		if err != nil {
			w.Logger.Error("Failed to resubscribe", "err", err)
		}
	}
}
1230054
func (i *Index) MergeChunks(s func([]bgzf.Chunk) []bgzf.Chunk) {
	if s == nil {
		return
	}
	for _, ref := range i.Refs {
		for b, bin := range ref.Bins {
			if !sort.IsSorted(byBeginOffset(bin.Chunks)) {
				sort.Sort(byBeginOffset(bin.Chunks))
			}
			ref.Bins[b].Chunks = s(bin.Chunks)
			if !sort.IsSorted(byBeginOffset(bin.Chunks)) {
				sort.Sort(byBeginOffset(bin.Chunks))
			}
		}
	}
}
685601
func pullSubcription(c context.Context, subscription, pubSubURL string) (*pubsub.PubsubMessage, func(), error) {
	service, err := createPubSubService(c, pubSubURL)
	if err != nil {
		return nil, nil, err
	}

	resp, err := service.Projects.Subscriptions.Pull(subscription, &pubsub.PullRequest{
		ReturnImmediately: true,
		MaxMessages:       1,
	}).Context(c).Do()
	if err != nil {
		return nil, nil, err
	}

	switch len(resp.ReceivedMessages) {
	case 0:
		return nil, nil, nil
	case 1:
		ackID := resp.ReceivedMessages[0].AckId
		ackCb := func() {
			_, err := service.Projects.Subscriptions.Acknowledge(subscription, &pubsub.AcknowledgeRequest{
				AckIds: []string{ackID},
			}).Context(c).Do()
			if err != nil {
				logging.Errorf(c, "Failed to acknowledge the message - %s", err)
			}
		}
		return resp.ReceivedMessages[0].Message, ackCb, nil
	default:
		panic(errors.New("received more than one message from PubSub while asking only one"))
	}
}
1328327
func (r *Request) fromChunk(chunk storage.Chunk) error {
	// for update chunk layout see Request definition

	chunkdata := chunk.Data()

	//deserialize the feed update portion
	if err := r.Update.binaryGet(chunkdata[:len(chunkdata)-signatureLength]); err != nil {
		return err
	}

	// Extract the signature
	var signature *Signature
	cursor := r.Update.binaryLength()
	sigdata := chunkdata[cursor : cursor+signatureLength]
	if len(sigdata) > 0 {
		signature = &Signature{}
		copy(signature[:], sigdata)
	}

	r.Signature = signature
	r.idAddr = chunk.Address()
	r.binaryData = chunkdata

	return nil

}