// ntfnHandler handles the logic to retrieve ZMQ notifications from the backing
// bitcoind connection.
//
// NOTE: This must be called as a goroutine.
426344
func (c *BitcoindClient) ntfnHandler() {
	defer c.wg.Done()

	for {
		select {
		case tx := <-c.zmqTxNtfns:
			if _, _, err := c.filterTx(tx, nil, true); err != nil {
				log.Errorf("Unable to filter transaction %v: %v",
					tx.TxHash(), err)
			}
		case newBlock := <-c.zmqBlockNtfns:
			// If the new block's previous hash matches the best
			// hash known to us, then the new block is the next
			// successor, so we'll update our best block to reflect
			// this and determine if this new block matches any of
			// our existing filters.
			c.bestBlockMtx.Lock()
			bestBlock := c.bestBlock
			c.bestBlockMtx.Unlock()
			if newBlock.Header.PrevBlock == bestBlock.Hash {
				newBlockHeight := bestBlock.Height + 1
				_, err := c.filterBlock(
					newBlock, newBlockHeight, true,
				)
				if err != nil {
					log.Errorf("Unable to filter block %v: %v",
						newBlock.BlockHash(), err)
					continue
				}

				// With the block succesfully filtered, we'll
				// make it our new best block.
				bestBlock.Hash = newBlock.BlockHash()
				bestBlock.Height = newBlockHeight
				bestBlock.Timestamp = newBlock.Header.Timestamp

				c.bestBlockMtx.Lock()
				c.bestBlock = bestBlock
				c.bestBlockMtx.Unlock()

				continue
			}

			// Otherwise, we've encountered a reorg.
			if err := c.reorg(bestBlock, newBlock); err != nil {
				log.Errorf("Unable to process chain reorg: %v",
					err)
			}
		case <-c.quit:
			return
		}
	}
}
744395
func (s *NotificationServer) TransactionNotifications() TransactionNotificationsClient {
	c := make(chan *TransactionNotifications)
	s.mu.Lock()
	s.transactions = append(s.transactions, c)
	s.mu.Unlock()
	return TransactionNotificationsClient{
		C:      c,
		server: s,
	}
}
11632
func (c *BitcoindConn) getCurrentNet() (wire.BitcoinNet, error) {
	hash, err := c.client.GetBlockHash(0)
	if err != nil {
		return 0, err
	}

	switch *hash {
	case *chaincfg.TestNet3Params.GenesisHash:
		return chaincfg.TestNet3Params.Net, nil
	case *chaincfg.RegressionNetParams.GenesisHash:
		return chaincfg.RegressionNetParams.Net, nil
	case *chaincfg.MainNetParams.GenesisHash:
		return chaincfg.MainNetParams.Net, nil
	default:
		return 0, fmt.Errorf("unknown network with genesis hash %v", hash)
	}
}
1049505
func (s *IdentityNotificationAttributes) SetHeadersInDeliveryNotificationsEnabled(v bool) *IdentityNotificationAttributes {
	s.HeadersInDeliveryNotificationsEnabled = &v
	return s
}
1264806
func (f *FSNotifications) QueueNotification(fn func()) {
	f.notificationMutex.RLock()
	defer f.notificationMutex.RUnlock()
	if f.notifications == nil {
		f.log.Warning("Ignoring notification, no available channel")
		return
	}
	f.notificationGroup.Add(1)
	f.notifications.In() <- fn
}
1330091
func (c *BitcoindClient) onRelevantTx(tx *wtxmgr.TxRecord,
	blockDetails *btcjson.BlockDetails) {

	block, err := parseBlock(blockDetails)
	if err != nil {
		log.Errorf("Unable to send onRelevantTx notification, failed "+
			"parse block: %v", err)
		return
	}

	select {
	case c.notificationQueue.ChanIn() <- RelevantTx{
		TxRecord: tx,
		Block:    block,
	}:
	case <-c.quit:
	}
}
1342587
func (n *NeutrinoNotifier) notificationDispatcher() {
	defer n.wg.Done()
out:
	for {
		select {
		case cancelMsg := <-n.notificationCancels:
			switch msg := cancelMsg.(type) {
			case *epochCancel:
				chainntnfs.Log.Infof("Cancelling epoch "+
					"notification, epoch_id=%v", msg.epochID)

				// First, we'll lookup the original
				// registration in order to stop the active
				// queue goroutine.
				reg := n.blockEpochClients[msg.epochID]
				reg.epochQueue.Stop()

				// Next, close the cancel channel for this
				// specific client, and wait for the client to
				// exit.
				close(n.blockEpochClients[msg.epochID].cancelChan)
				n.blockEpochClients[msg.epochID].wg.Wait()

				// Once the client has exited, we can then
				// safely close the channel used to send epoch
				// notifications, in order to notify any
				// listeners that the intent has been
				// cancelled.
				close(n.blockEpochClients[msg.epochID].epochChan)
				delete(n.blockEpochClients, msg.epochID)
			}

		case registerMsg := <-n.notificationRegistry:
			switch msg := registerMsg.(type) {
			case *chainntnfs.HistoricalConfDispatch:
				// We'll start a historical rescan chain of the
				// chain asynchronously to prevent blocking
				// potentially long rescans.
				n.wg.Add(1)
				go func() {
					defer n.wg.Done()

					confDetails, err := n.historicalConfDetails(
						msg.ConfRequest,
						msg.StartHeight, msg.EndHeight,
					)
					if err != nil {
						chainntnfs.Log.Error(err)
						return
					}

					// If the historical dispatch finished
					// without error, we will invoke
					// UpdateConfDetails even if none were
					// found. This allows the notifier to
					// begin safely updating the height hint
					// cache at tip, since any pending
					// rescans have now completed.
					err = n.txNotifier.UpdateConfDetails(
						msg.ConfRequest, confDetails,
					)
					if err != nil {
						chainntnfs.Log.Error(err)
					}
				}()

			case *blockEpochRegistration:
				chainntnfs.Log.Infof("New block epoch subscription")

				n.blockEpochClients[msg.epochID] = msg

				// If the client did not provide their best
				// known block, then we'll immediately dispatch
				// a notification for the current tip.
				if msg.bestBlock == nil {
					n.notifyBlockEpochClient(
						msg, n.bestBlock.Height,
						n.bestBlock.Hash,
					)

					msg.errorChan <- nil
					continue
				}

				// Otherwise, we'll attempt to deliver the
				// backlog of notifications from their best
				// known block.
				n.bestBlockMtx.Lock()
				bestHeight := n.bestBlock.Height
				n.bestBlockMtx.Unlock()

				missedBlocks, err := chainntnfs.GetClientMissedBlocks(
					n.chainConn, msg.bestBlock, bestHeight,
					false,
				)
				if err != nil {
					msg.errorChan <- err
					continue
				}

				for _, block := range missedBlocks {
					n.notifyBlockEpochClient(
						msg, block.Height, block.Hash,
					)
				}

				msg.errorChan <- nil

			case *rescanFilterUpdate:
				err := n.chainView.Update(msg.updateOptions...)
				if err != nil {
					chainntnfs.Log.Errorf("Unable to "+
						"update rescan filter: %v", err)
				}
				msg.errChan <- err
			}

		case item := <-n.chainUpdates.ChanOut():
			update := item.(*filteredBlock)
			if update.connect {
				n.bestBlockMtx.Lock()
				// Since neutrino has no way of knowing what
				// height to rewind to in the case of a reorged
				// best known height, there is no point in
				// checking that the previous hash matches the
				// the hash from our best known height the way
				// the other notifiers do when they receive
				// a new connected block. Therefore, we just
				// compare the heights.
				if update.height != uint32(n.bestBlock.Height+1) {
					// Handle the case where the notifier
					// missed some blocks from its chain
					// backend
					chainntnfs.Log.Infof("Missed blocks, " +
						"attempting to catch up")

					_, missedBlocks, err :=
						chainntnfs.HandleMissedBlocks(
							n.chainConn,
							n.txNotifier,
							n.bestBlock,
							int32(update.height),
							false,
						)
					if err != nil {
						chainntnfs.Log.Error(err)
						n.bestBlockMtx.Unlock()
						continue
					}

					for _, block := range missedBlocks {
						filteredBlock, err :=
							n.getFilteredBlock(block)
						if err != nil {
							chainntnfs.Log.Error(err)
							n.bestBlockMtx.Unlock()
							continue out
						}
						err = n.handleBlockConnected(filteredBlock)
						if err != nil {
							chainntnfs.Log.Error(err)
							n.bestBlockMtx.Unlock()
							continue out
						}
					}

				}

				err := n.handleBlockConnected(update)
				if err != nil {
					chainntnfs.Log.Error(err)
				}

				n.bestBlockMtx.Unlock()
				continue
			}

			n.bestBlockMtx.Lock()
			if update.height != uint32(n.bestBlock.Height) {
				chainntnfs.Log.Infof("Missed disconnected " +
					"blocks, attempting to catch up")
			}
			newBestBlock, err := chainntnfs.RewindChain(
				n.chainConn, n.txNotifier, n.bestBlock,
				int32(update.height-1),
			)
			if err != nil {
				chainntnfs.Log.Errorf("Unable to rewind chain "+
					"from height %d to height %d: %v",
					n.bestBlock.Height, update.height-1, err)
			}

			// Set the bestHeight here in case a chain rewind
			// partially completed.
			n.bestBlock = newBestBlock
			n.bestBlockMtx.Unlock()

		case txUpdate := <-n.txUpdates.ChanOut():
			// A new relevant transaction notification has been
			// received from the backend. We'll attempt to process
			// it to determine if it fulfills any outstanding
			// confirmation and/or spend requests and dispatch
			// notifications for them.
			update := txUpdate.(*relevantTx)
			err := n.txNotifier.ProcessRelevantSpendTx(
				update.tx, uint32(update.details.Height),
			)
			if err != nil {
				chainntnfs.Log.Errorf("Unable to process "+
					"transaction %v: %v", update.tx.Hash(),
					err)
			}

		case err := <-n.rescanErr:
			chainntnfs.Log.Errorf("Error during rescan: %v", err)

		case <-n.quit:
			return

		}
	}
}
112373
func (c *Client) reregisterNtfns() error {
	// Nothing to do if the caller is not interested in notifications.
	if c.ntfnHandlers == nil {
		return nil
	}

	// In order to avoid holding the lock on the notification state for the
	// entire time of the potentially long running RPCs issued below, make a
	// copy of it and work from that.
	//
	// Also, other commands will be running concurrently which could modify
	// the notification state (while not under the lock of course) which
	// also register it with the remote RPC server, so this prevents double
	// registrations.
	c.ntfnStateLock.Lock()
	stateCopy := c.ntfnState.Copy()
	c.ntfnStateLock.Unlock()

	// Reregister notifyblocks if needed.
	if stateCopy.notifyBlocks {
		log.Debugf("Reregistering [notifyblocks]")
		if err := c.NotifyBlocks(); err != nil {
			return err
		}
	}

	// Reregister notifynewtransactions if needed.
	if stateCopy.notifyNewTx || stateCopy.notifyNewTxVerbose {
		log.Debugf("Reregistering [notifynewtransactions] (verbose=%v)",
			stateCopy.notifyNewTxVerbose)
		err := c.NotifyNewTransactions(stateCopy.notifyNewTxVerbose)
		if err != nil {
			return err
		}
	}

	// Reregister the combination of all previously registered notifyspent
	// outpoints in one command if needed.
	nslen := len(stateCopy.notifySpent)
	if nslen > 0 {
		outpoints := make([]btcjson.OutPoint, 0, nslen)
		for op := range stateCopy.notifySpent {
			outpoints = append(outpoints, op)
		}
		log.Debugf("Reregistering [notifyspent] outpoints: %v", outpoints)
		if err := c.notifySpentInternal(outpoints).Receive(); err != nil {
			return err
		}
	}

	// Reregister the combination of all previously registered
	// notifyreceived addresses in one command if needed.
	nrlen := len(stateCopy.notifyReceived)
	if nrlen > 0 {
		addresses := make([]string, 0, nrlen)
		for addr := range stateCopy.notifyReceived {
			addresses = append(addresses, addr)
		}
		log.Debugf("Reregistering [notifyreceived] addresses: %v", addresses)
		if err := c.notifyReceivedInternal(addresses).Receive(); err != nil {
			return err
		}
	}

	return nil
}
593726
func (c *Client) trackRegisteredNtfns(cmd interface{}) {
	// Nothing to do if the caller is not interested in notifications.
	if c.ntfnHandlers == nil {
		return
	}

	c.ntfnStateLock.Lock()
	defer c.ntfnStateLock.Unlock()

	switch bcmd := cmd.(type) {
	case *btcjson.NotifyBlocksCmd:
		c.ntfnState.notifyBlocks = true

	case *btcjson.NotifyNewTransactionsCmd:
		if bcmd.Verbose != nil && *bcmd.Verbose {
			c.ntfnState.notifyNewTxVerbose = true
		} else {
			c.ntfnState.notifyNewTx = true

		}

	case *btcjson.NotifySpentCmd:
		for _, op := range bcmd.OutPoints {
			c.ntfnState.notifySpent[op] = struct{}{}
		}

	case *btcjson.NotifyReceivedCmd:
		for _, addr := range bcmd.Addresses {
			c.ntfnState.notifyReceived[addr] = struct{}{}
		}
	}
}
262240
func (s *CreateFunctionInput) SetHandler(v string) *CreateFunctionInput {
	s.Handler = &v
	return s
}
491983
func (s *rpcServer) NotifyNewTransactions(txns []*mempool.TxDesc) {
	for _, txD := range txns {
		// Notify websocket clients about mempool transactions.
		s.ntfnMgr.NotifyMempoolTx(txD.Tx, true)

		// Potentially notify any getblocktemplate long poll clients
		// about stale block templates due to the new transaction.
		s.gbtWorkState.NotifyMempoolTx(s.cfg.TxMemPool.LastUpdated())
	}
}
1074141
func newRedeemingTxNotification(txHex string, index int, block *btcutil.Block) ([]byte, error) {
	// Create and marshal the notification.
	ntfn := btcjson.NewRedeemingTxNtfn(txHex, blockDetails(block, index))
	return btcjson.MarshalCmd(nil, ntfn)
}
642528
func (s *FunctionConfiguration) SetHandler(v string) *FunctionConfiguration {
	s.Handler = &v
	return s
}
725044
func NotificationHandlers(ntfn rpcclient.NotificationHandlers) RescanOption {
	return func(ro *rescanOptions) {
		ro.ntfn = ntfn
	}
}
548520
func errHandler(err error) handlerFunc {
	return func(w http.ResponseWriter, r *http.Request) error {
		return err
	}
}
785468
func NewReceivedPacketHandler(
	rttStats *congestion.RTTStats,
	logger utils.Logger,
	version protocol.VersionNumber,
) ReceivedPacketHandler {
	return &receivedPacketHandler{
		initialPackets:   newReceivedPacketTracker(rttStats, logger, version),
		handshakePackets: newReceivedPacketTracker(rttStats, logger, version),
		oneRTTPackets:    newReceivedPacketTracker(rttStats, logger, version),
	}
}
1036889
func (s *LoadBalancerTlsCertificate) SetNotAfter(v time.Time) *LoadBalancerTlsCertificate {
	s.NotAfter = &v
	return s
}