// writes a record in the subscriptions file.
// store's lock is held on entry.
9744
func (ss *FileSubStore) writeRecord(w io.Writer, recType recordType, rec record) error {
	var err error
	totalSize := 0
	recSize := rec.Size()

	var bwBuf *bufio.Writer
	needsUnlock := false

	if w == nil {
		if err := ss.lockFile(); err != nil {
			return err
		}
		needsUnlock = true
		if ss.bw != nil {
			bwBuf = ss.bw.buf
			// If we are using the buffer writer on this call, and the buffer is
			// not already at the max size...
			if bwBuf != nil && ss.bw.bufSize != ss.opts.BufferSize {
				// Check if record fits
				required := recSize + recordHeaderSize
				if required > bwBuf.Available() {
					ss.writer, err = ss.bw.expand(ss.file.handle, required)
					if err != nil {
						ss.fm.unlockFile(ss.file)
						return err
					}
					bwBuf = ss.bw.buf
				}
			}
		}
		w = ss.writer
	}
	ss.tmpSubBuf, totalSize, err = writeRecord(w, ss.tmpSubBuf, recType, rec, recSize, ss.crcTable)
	if err != nil {
		if needsUnlock {
			ss.fm.unlockFile(ss.file)
		}
		return err
	}
	if bwBuf != nil && ss.bw.shrinkReq {
		ss.bw.checkShrinkRequest()
	}
	// Indicate that we wrote something to the buffer/file
	ss.activity = true
	switch recType {
	case subRecNew:
		ss.numRecs++
	case subRecMsg:
		ss.numRecs++
	case subRecAck:
		// An ack makes the message record free space
		ss.delRecs++
	case subRecUpdate:
		ss.numRecs++
		// An update makes the old record free space
		ss.delRecs++
	case subRecDel:
		ss.delRecs++
	default:
		panic(fmt.Errorf("record type %v unknown", recType))
	}
	ss.fileSize += int64(totalSize)
	if needsUnlock {
		ss.fm.unlockFile(ss.file)
	}
	return nil
}
552760
func (s *MemoryUserStore) Store(user *msp.UserData) error {
	s.store[user.ID+"@"+user.MSPID] = user.EnrollmentCertificate
	return nil
}
440672
func (f *FileStorage) Close() error {
	f.subscription.Close()
	return f.file.Close()
}
611859
func (o *os) Subscribe(ctx context.Context, h Handler, types ...string) error {
	req := &ev.StreamRequest{}
	for _, typ := range types {
		req.Types = append(req.Types, typ)
	}

	stream, err := o.cl.Stream(ctx, req)
	if err != nil {
		return err
	}

	for {
		rsp, err := stream.Recv()
		if err != nil {
			return err
		}
		h(toRecord(rsp.Record))
	}

	return nil
}
1044339
func (r *addrsRecord) flush(write ds.Write) (err error) {
	key := addrBookBase.ChildString(b32.RawStdEncoding.EncodeToString([]byte(r.Id.ID)))
	if len(r.Addrs) == 0 {
		if err = write.Delete(key); err == nil {
			r.dirty = false
		}
		return err
	}

	data, err := r.Marshal()
	if err != nil {
		return err
	}
	if err = write.Put(key, data); err != nil {
		return err
	}
	// write succeeded; record is no longer dirty.
	r.dirty = false
	return nil
}
380491
func (rs *RaftStorage) Store(entries []raftpb.Entry, hardstate raftpb.HardState, snapshot raftpb.Snapshot) error {
	if err := rs.wal.Save(hardstate, entries); err != nil {
		return err
	}

	if !raft.IsEmptySnap(snapshot) {
		if err := rs.saveSnap(snapshot); err != nil {
			return err
		}

		if err := rs.ram.ApplySnapshot(snapshot); err != nil {
			if err == raft.ErrSnapOutOfDate {
				rs.lg.Warnf("Attempted to apply out-of-date snapshot at Term %d and Index %d",
					snapshot.Metadata.Term, snapshot.Metadata.Index)
			} else {
				rs.lg.Fatalf("Unexpected programming error: %s", err)
			}
		}
	}

	if err := rs.ram.Append(entries); err != nil {
		return err
	}

	return nil
}
864068
func (s *Store) Close() (lastIdx uint64, err error) {
	// Notify goroutines of closing and wait until they finish.
	close(s.closing)
	s.wg.Wait()
	s.mu.Lock()
	defer s.mu.Unlock()

	for _, l := range s.subscribers {
		for el := l.Front(); el != nil; el = el.Next() {
			go el.Value.(*subscription).Close()
		}
	}
	if s.raft != nil {
		s.raft.Shutdown().Error()
		lastIdx = s.raft.LastIndex()
		s.raft = nil
	}
	if s.transport != nil {
		s.transport.Close()
		s.transport = nil
	}
	if s.stableStore != nil {
		s.stableStore.Close()
		s.stableStore = nil
	}

	return lastIdx, nil
}
1024061
func (b *messageBuffer) store(m *message) {
	b.mu.Lock()
	defer b.mu.Unlock()

	if len(b.backlog) == 0 {
		select {
		case b.c <- m:
			return
		default:
		}
	}
	b.backlog = append(b.backlog, m)
}
442525
func (s *session) recordCommited(rec *sessionRecord) {
	if rec.has(recJournalNum) {
		s.stJournalNum = rec.journalNum
	}

	if rec.has(recPrevJournalNum) {
		s.stPrevJournalNum = rec.prevJournalNum
	}

	if rec.has(recSeqNum) {
		s.stSeqNum = rec.seqNum
	}

	for _, r := range rec.compPtrs {
		s.setCompPtr(r.level, internalKey(r.ikey))
	}
}
341156
func (ss *SQLSubStore) Flush() error {
	ss.Lock()
	err := ss.flush()
	ss.Unlock()
	return err
}
1286986
func (ss *FileSubStore) CreateSub(sub *spb.SubState) error {
	// Check if we can create the subscription (check limits and update
	// subscription count)
	ss.Lock()
	defer ss.Unlock()
	if err := ss.createSub(sub); err != nil {
		return err
	}
	if err := ss.writeRecord(nil, subRecNew, sub); err != nil {
		delete(ss.subs, sub.ID)
		return err
	}
	// We need to get a copy of the passed sub, we can't hold a reference
	// to it.
	csub := *sub
	s := &subscription{sub: &csub, seqnos: make(map[uint64]struct{})}
	ss.subs[sub.ID] = s
	return nil
}
152263
func (p *PeekingSource) Record() (interface{}, error) {
	p.mu.RLock()
	var rec interface{}
	if p.rec == nil {
		defer p.mu.RUnlock()
		return p.Source.Record()
	} else {
		// Exchange read lock for write lock and recheck p.rec
		p.mu.RUnlock()
		p.mu.Lock()
		if p.rec == nil {
			p.mu.Unlock()
			return p.Record()
		}
		rec = p.rec
		p.rec = nil
		p.mu.Unlock()
		return rec, nil
	}
}
780948
func subscribe(p *Proxy) {
	lumber.Trace("Adding proxy to subscribers...")

	mutex.Lock()
	subscribers[p.id] = p
	mutex.Unlock()
}
589673
func (ss *FileSubStore) Flush() error {
	ss.Lock()
	err := ss.lockFile()
	if err == nil {
		err = ss.flush()
		ss.fm.unlockFile(ss.file)
	}
	ss.Unlock()
	return err
}
987169
func (o *FileSystemSaver) Save(store uuid.Store) {

	if store.Timestamp >= o.Timestamp {
		err := o.openAndDo(o.encode, &store)
		if err == nil {
			if o.Report {
				o.Println("file system saver saved", store)
			}
		}
		o.Timestamp = store.Add(o.Duration)
	}
}
136856
func (ss *FileSubStore) AckSeqPending(subid, seqno uint64) error {
	ss.Lock()
	ss.updateSub.ID, ss.updateSub.Seqno = subid, seqno
	if err := ss.writeRecord(nil, subRecAck, &ss.updateSub); err != nil {
		ss.Unlock()
		return err
	}
	si := ss.subs[subid]
	if si != nil {
		s := si.(*subscription)
		delete(s.seqnos, seqno)
		// Test if we should compact
		if ss.shouldCompact() {
			ss.fm.closeFileIfOpened(ss.file)
			ss.compact(ss.file.name)
		}
	}
	ss.Unlock()
	return nil
}
861596
func (c Client) AddSubscription(s contract.Subscription) (string, error) {
	conn := c.Pool.Get()
	defer conn.Close()

	err := addSubscription(conn, &s)
	if err != nil {
		return "", err
	}

	return s.ID, nil
}