// NewFilteredBlockEvent creates a new FilteredBlockEvent
2530
func NewFilteredBlockEvent(fblock *pb.FilteredBlock, sourceURL string) *fab.FilteredBlockEvent {
	return &fab.FilteredBlockEvent{
		FilteredBlock: fblock,
		SourceURL:     sourceURL,
	}
}
299684
func (m *wsNotificationManager) notifyFilteredBlockConnected(clients map[chan struct{}]*wsClient,
	block *btcutil.Block) {

	// Create the common portion of the notification that is the same for
	// every client.
	var w bytes.Buffer
	err := block.MsgBlock().Header.Serialize(&w)
	if err != nil {
		rpcsLog.Errorf("Failed to serialize header for filtered block "+
			"connected notification: %v", err)
		return
	}
	ntfn := btcjson.NewFilteredBlockConnectedNtfn(block.Height(),
		hex.EncodeToString(w.Bytes()), nil)

	// Search for relevant transactions for each client and save them
	// serialized in hex encoding for the notification.
	subscribedTxs := make(map[chan struct{}][]string)
	for _, tx := range block.Transactions() {
		var txHex string
		for quitChan := range m.subscribedClients(tx, clients) {
			if txHex == "" {
				txHex = txHexString(tx.MsgTx())
			}
			subscribedTxs[quitChan] = append(subscribedTxs[quitChan], txHex)
		}
	}
	for quitChan, wsc := range clients {
		// Add all discovered transactions for this client. For clients
		// that have no new-style filter, add the empty string slice.
		ntfn.SubscribedTxs = subscribedTxs[quitChan]

		// Marshal and queue notification.
		marshalledJSON, err := btcjson.MarshalCmd(nil, ntfn)
		if err != nil {
			rpcsLog.Errorf("Failed to marshal filtered block "+
				"connected notification: %v", err)
			return
		}
		wsc.QueueNotification(marshalledJSON)
	}
}
1212983
func NewFatalDisconnectedEvent(cause error) *DisconnectedEvent {
	return &DisconnectedEvent{Err: &disconnectedError{cause: cause, fatal: true}}
}
257954
func (*wsNotificationManager) notifyFilteredBlockDisconnected(clients map[chan struct{}]*wsClient,
	block *btcutil.Block) {
	// Skip notification creation if no clients have requested block
	// connected/disconnected notifications.
	if len(clients) == 0 {
		return
	}

	// Notify interested websocket clients about the disconnected block.
	var w bytes.Buffer
	err := block.MsgBlock().Header.Serialize(&w)
	if err != nil {
		rpcsLog.Errorf("Failed to serialize header for filtered block "+
			"disconnected notification: %v", err)
		return
	}
	ntfn := btcjson.NewFilteredBlockDisconnectedNtfn(block.Height(),
		hex.EncodeToString(w.Bytes()))
	marshalledJSON, err := btcjson.MarshalCmd(nil, ntfn)
	if err != nil {
		rpcsLog.Errorf("Failed to marshal filtered block disconnected "+
			"notification: %v", err)
		return
	}
	for _, wsc := range clients {
		wsc.QueueNotification(marshalledJSON)
	}
}
863726
func (ref *EventClientRef) RegisterBlockEvent(filter ...fab.BlockFilter) (fab.Registration, <-chan *fab.BlockEvent, error) {
	service, err := ref.get()
	if err != nil {
		return nil, nil, err
	}
	return service.RegisterBlockEvent(filter...)
}
389544
func (f *EventSourceObjectSpamFilter) Filter(event *v1.Event) bool {
	var record spamRecord

	// controls our cached information about this event (source+object)
	eventKey := getSpamKey(event)

	// do we have a record of similar events in our cache?
	f.Lock()
	defer f.Unlock()
	value, found := f.cache.Get(eventKey)
	if found {
		record = value.(spamRecord)
	}

	// verify we have a rate limiter for this record
	if record.rateLimiter == nil {
		record.rateLimiter = flowcontrol.NewTokenBucketRateLimiterWithClock(f.qps, f.burst, f.clock)
	}

	// ensure we have available rate
	filter := !record.rateLimiter.TryAccept()

	// update the cache
	f.cache.Add(eventKey, record)

	return filter
}
104416
func (ed *Dispatcher) HandleFilteredBlock(fblock *pb.FilteredBlock, sourceURL string) {
	logger.Debugf("Handling filtered block event - Block #%d", fblock.Number)

	if err := ed.updateLastBlockNum(fblock.Number); err != nil {
		logger.Error(err.Error())
		return
	}

	if ed.updateLastBlockInfoOnly {
		ed.updateLastBlockInfoOnly = false
		return
	}

	logger.Debug("Publishing filtered block event...")
	ed.publishFilteredBlockEvents(fblock, sourceURL)
}
1351940
func (s *Selection) NextFiltered(selector string) *Selection {
	return filterAndPush(s, getSiblingNodes(s.Nodes, siblingNext, nil, nil), compileMatcher(selector))
}
811379
func newBlockNode(blockHeader *wire.BlockHeader, parent *blockNode) *blockNode {
	var node blockNode
	initBlockNode(&node, blockHeader, parent)
	return &node
}
1100722
func NewEventSourceObjectSpamFilter(lruCacheSize, burst int, qps float32, clock clock.Clock) *EventSourceObjectSpamFilter {
	return &EventSourceObjectSpamFilter{
		cache: lru.New(lruCacheSize),
		burst: burst,
		qps:   qps,
		clock: clock,
	}
}
105896
func AddEventWatch(filter EventFilter, userdata interface{}) EventWatchHandle {
	context := newEventFilterCallbackContext(filter, userdata)
	C.addEventWatch(context.cptr())
	return context.handle
}
1041961
func newBlockEventQueue() *blockEventQueue {
	b := &blockEventQueue{
		newBlocks:   make(chan *FilteredBlock),
		staleBlocks: make(chan *FilteredBlock),
		quit:        make(chan struct{}),
	}
	b.queueCond = sync.NewCond(&b.queueMtx)

	return b
}
696122
func (be *BlockExecution) StreamEvents() StreamEvents {
	var ses StreamEvents
	ses = append(ses, &StreamEvent{
		BeginBlock: &BeginBlock{
			Height: be.Height,
			Header: be.Header,
		},
	})
	for _, txe := range be.TxExecutions {
		ses = append(ses, txe.StreamEvents()...)
	}
	return append(ses, &StreamEvent{
		EndBlock: &EndBlock{
			Height: be.Height,
		},
	})
}
335262
func newBlockCallback(n *Node) em.EventCallbackFunc {
	return func(metric *em.EventMetric, data interface{}) {
		block := data.(tmtypes.TMEventData).(tmtypes.EventDataNewBlockHeader).Header

		n.Height = block.Height
		n.logger.Info("new block", "height", block.Height, "numTxs", block.NumTxs)

		if n.blockCh != nil {
			n.blockCh <- block
		}
	}
}
269360
func (s *Selection) ParentsFiltered(selector string) *Selection {
	return filterAndPush(s, getParentsNodes(s.Nodes, nil, nil), compileMatcher(selector))
}
368672
func (s *InputTemplate) SetDeblockFilter(v string) *InputTemplate {
	s.DeblockFilter = &v
	return s
}
747386
func NewFilteredTxWithCCEvent(txID, ccID, event string) *pb.FilteredTransaction {
	return &pb.FilteredTransaction{
		Txid: txID,
		Data: &pb.FilteredTransaction_TransactionActions{
			TransactionActions: &pb.FilteredTransactionActions{
				ChaincodeActions: []*pb.FilteredChaincodeAction{
					{
						ChaincodeEvent: &pb.ChaincodeEvent{
							ChaincodeId: ccID,
							EventName:   event,
							TxId:        txID,
						},
					},
				},
			},
		},
	}
}