// NewBlockchainReactor returns new reactor instance.
482159
func NewBlockchainReactor(state sm.State, blockExec *sm.BlockExecutor, store *BlockStore,
	fastSync bool) *BlockchainReactor {

	if state.LastBlockHeight != store.Height() {
		panic(fmt.Sprintf("state (%v) and store (%v) height mismatch", state.LastBlockHeight,
			store.Height()))
	}

	requestsCh := make(chan BlockRequest, maxTotalRequesters)

	const capacity = 1000                      // must be bigger than peers count
	errorsCh := make(chan peerError, capacity) // so we don't block in #Receive#pool.AddBlock

	pool := NewBlockPool(
		store.Height()+1,
		requestsCh,
		errorsCh,
	)

	bcR := &BlockchainReactor{
		initialState: state,
		blockExec:    blockExec,
		store:        store,
		pool:         pool,
		fastSync:     fastSync,
		requestsCh:   requestsCh,
		errorsCh:     errorsCh,
	}
	bcR.BaseReactor = *p2p.NewBaseReactor("BlockchainReactor", bcR)
	return bcR
}
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
		}
	}
}
213626
func NewMockLedger(eventFactory EventFactory, sourceURL string) *MockLedger {
	return &MockLedger{
		eventFactory:  eventFactory,
		blockProducer: NewBlockProducer(),
		sourceURL:     sourceURL,
	}
}
186912
func NewBlockCapability(
	mode csi.VolumeCapability_AccessMode_Mode) *csi.VolumeCapability {

	return &csi.VolumeCapability{
		AccessMode: &csi.VolumeCapability_AccessMode{
			Mode: mode,
		},
		AccessType: &csi.VolumeCapability_Block{
			Block: &csi.VolumeCapability_BlockVolume{},
		},
	}
}
980082
func (s *Store) NewIterateRecentBlocks() *BlockIterator {
	s.mtx.RLock()
	defer s.mtx.RUnlock()

	return s.recent.iter(s)
}
840130
func newChainWatcher(cfg chainWatcherConfig) (*chainWatcher, error) {
	// In order to be able to detect the nature of a potential channel
	// closure we'll need to reconstruct the state hint bytes used to
	// obfuscate the commitment state number encoded in the lock time and
	// sequence fields.
	var stateHint [lnwallet.StateHintSize]byte
	chanState := cfg.chanState
	if chanState.IsInitiator {
		stateHint = lnwallet.DeriveStateHintObfuscator(
			chanState.LocalChanCfg.PaymentBasePoint.PubKey,
			chanState.RemoteChanCfg.PaymentBasePoint.PubKey,
		)
	} else {
		stateHint = lnwallet.DeriveStateHintObfuscator(
			chanState.RemoteChanCfg.PaymentBasePoint.PubKey,
			chanState.LocalChanCfg.PaymentBasePoint.PubKey,
		)
	}

	return &chainWatcher{
		cfg:                 cfg,
		stateHintObfuscator: stateHint,
		quit:                make(chan struct{}),
		clientSubscriptions: make(map[uint64]*ChainEventSubscription),
	}, nil
}
1160512
func NewChainProcessor(processors ...Processor) ProcessorFunc {
	return func(in string) (string, bool) {
		out := in
		ok := true
		for _, processor := range processors {
			out, ok = processor.Process(out)
			if !ok {
				return "", false
			}
		}
		return out, ok
	}
}
279627
func NewKinesisReactor(reactor KinesisReactor,
	kinesisStream gocf.Stringable,
	startingPosition string,
	batchSize int64,
	additionalLambdaPermissions []sparta.IAMRolePrivilege) (*sparta.LambdaAWSInfo, error) {

	reactorLambda := func(ctx context.Context, kinesisEvent awsLambdaEvents.KinesisEvent) (interface{}, error) {
		return reactor.OnKinesisMessage(ctx, kinesisEvent)
	}

	lambdaFn, lambdaFnErr := sparta.NewAWSLambda(reactorName(reactor),
		reactorLambda,
		sparta.IAMRoleDefinition{})
	if lambdaFnErr != nil {
		return nil, errors.Wrapf(lambdaFnErr, "attempting to create reactor")
	}

	lambdaFn.EventSourceMappings = append(lambdaFn.EventSourceMappings,
		&sparta.EventSourceMapping{
			EventSourceArn:   kinesisStream,
			StartingPosition: startingPosition,
			BatchSize:        batchSize,
		})
	if len(additionalLambdaPermissions) != 0 {
		lambdaFn.RoleDefinition.Privileges = additionalLambdaPermissions
	}
	return lambdaFn, nil
}
877566
func (p *BlockProducer) NewBlock(channelID string, transactions ...*TxInfo) *cb.Block {
	block := NewBlock(channelID, transactions...)
	block.Header.Number = p.blockNum
	atomic.AddUint64(&p.blockNum, 1)
	return block
}
559837
func newInsertIterator(chain types.Blocks, results <-chan error, validator Validator) *insertIterator {
	return &insertIterator{
		chain:     chain,
		results:   results,
		errors:    make([]error, 0, len(chain)),
		index:     -1,
		validator: validator,
	}
}
1126559
func NewBlocksProvider(chainID string, client streamClient, gossip GossipServiceAdapter, mcs api.MessageCryptoService) BlocksProvider {
	return &blocksProviderImpl{
		chainID:              chainID,
		client:               client,
		gossip:               gossip,
		mcs:                  mcs,
		wrongStatusThreshold: wrongStatusThreshold,
	}
}
207323
func (l *MockLedger) NewBlock(channelID string, transactions ...*TxInfo) Block {
	l.Lock()
	defer l.Unlock()
	block := NewBlockWrapper(l.blockProducer.NewBlock(channelID, transactions...))
	l.Store(block)
	return block
}
1251641
func newBlock(seqNum uint64, previousHash []byte) *common.Block {
	block := &common.Block{}
	block.Header = &common.BlockHeader{}
	block.Header.Number = seqNum
	block.Header.PreviousHash = previousHash
	block.Data = &common.BlockData{}

	var metadataContents [][]byte
	for i := 0; i < len(common.BlockMetadataIndex_name); i++ {
		metadataContents = append(metadataContents, []byte{})
	}
	block.Metadata = &common.BlockMetadata{Metadata: metadataContents}

	return block
}
849272
func newActiveChannelArbitrator(channel *channeldb.OpenChannel,
	c *ChainArbitrator, chanEvents *ChainEventSubscription) (*ChannelArbitrator, error) {

	log.Tracef("Creating ChannelArbitrator for ChannelPoint(%v)",
		channel.FundingOutpoint)

	// We'll start by registering for a block epoch notifications so this
	// channel can keep track of the current state of the main chain.
	//
	// TODO(roasbeef): fetch best height (or pass in) so can ensure block
	// epoch delivers all the notifications to
	//
	// TODO(roasbeef): instead 1 block epoch that multi-plexes to the rest?
	//  * reduces the number of goroutines
	blockEpoch, err := c.cfg.Notifier.RegisterBlockEpochNtfn(nil)
	if err != nil {
		return nil, err
	}

	chanPoint := channel.FundingOutpoint

	// Next we'll create the matching configuration struct that contains
	// all interfaces and methods the arbitrator needs to do its job.
	arbCfg := ChannelArbitratorConfig{
		ChanPoint:   chanPoint,
		ShortChanID: channel.ShortChanID(),
		BlockEpochs: blockEpoch,
		ForceCloseChan: func() (*lnwallet.LocalForceCloseSummary, error) {
			// First, we mark the channel as borked, this ensure
			// that no new state transitions can happen, and also
			// that the link won't be loaded into the switch.
			if err := channel.MarkBorked(); err != nil {
				return nil, err
			}

			// With the channel marked as borked, we'll now remove
			// the link from the switch if its there. If the link
			// is active, then this method will block until it
			// exits.
			if err := c.cfg.MarkLinkInactive(chanPoint); err != nil {
				log.Errorf("unable to mark link inactive: %v", err)
			}

			// Now that we know the link can't mutate the channel
			// state, we'll read the channel from disk the target
			// channel according to its channel point.
			channel, err := c.chanSource.FetchChannel(chanPoint)
			if err != nil {
				return nil, err
			}

			// Finally, we'll force close the channel completing
			// the force close workflow.
			chanMachine, err := lnwallet.NewLightningChannel(
				c.cfg.Signer, c.cfg.PreimageDB, channel, nil,
			)
			if err != nil {
				return nil, err
			}
			return chanMachine.ForceClose()
		},
		MarkCommitmentBroadcasted: channel.MarkCommitmentBroadcasted,
		MarkChannelClosed: func(summary *channeldb.ChannelCloseSummary) error {
			if err := channel.CloseChannel(summary); err != nil {
				return err
			}
			c.cfg.NotifyClosedChannel(summary.ChanPoint)
			return nil
		},
		IsPendingClose:        false,
		ChainArbitratorConfig: c.cfg,
		ChainEvents:           chanEvents,
	}

	// The final component needed is an arbitrator log that the arbitrator
	// will use to keep track of its internal state using a backed
	// persistent log.
	//
	// TODO(roasbeef); abstraction leak...
	//  * rework: adaptor method to set log scope w/ factory func
	chanLog, err := newBoltArbitratorLog(
		c.chanSource.DB, arbCfg, c.cfg.ChainHash, chanPoint,
	)
	if err != nil {
		blockEpoch.Cancel()
		return nil, err
	}

	arbCfg.MarkChannelResolved = func() error {
		return c.resolveContract(chanPoint, chanLog)
	}

	return NewChannelArbitrator(
		arbCfg, channel.LocalCommitment.Htlcs, chanLog,
	), nil
}
956343
func NewCloudWatchReactor(reactor CloudWatchReactor,
	subscriptions map[string]sparta.CloudWatchEventsRule,
	additionalLambdaPermissions []sparta.IAMRolePrivilege) (*sparta.LambdaAWSInfo, error) {
	if len(subscriptions) <= 0 {
		return nil, errors.Errorf("CloudWatchLogs subscription map must not be empty")
	}

	reactorLambda := func(ctx context.Context, cwLogs awsLambdaEvents.CloudwatchLogsEvent) (interface{}, error) {
		return reactor.OnCloudWatchMessage(ctx, cwLogs)
	}
	lambdaFn, lambdaFnErr := sparta.NewAWSLambda(reactorName(reactor),
		reactorLambda,
		sparta.IAMRoleDefinition{})
	if lambdaFnErr != nil {
		return nil, errors.Wrapf(lambdaFnErr, "attempting to create reactor")
	}
	cloudWatchEventsPermission := sparta.CloudWatchEventsPermission{}
	cloudWatchEventsPermission.Rules = make(map[string]sparta.CloudWatchEventsRule)
	for eachRuleName, eachRule := range subscriptions {
		cloudWatchEventsPermission.Rules[eachRuleName] = eachRule
	}
	lambdaFn.Permissions = append(lambdaFn.Permissions, cloudWatchEventsPermission)

	if len(additionalLambdaPermissions) != 0 {
		lambdaFn.RoleDefinition.Privileges = additionalLambdaPermissions
	}
	return lambdaFn, nil
}
57024
func NewSNSReactor(reactor SNSReactor,
	snsTopic gocf.Stringable,
	additionalLambdaPermissions []sparta.IAMRolePrivilege) (*sparta.LambdaAWSInfo, error) {

	reactorLambda := func(ctx context.Context, snsEvent awsLambdaEvents.SNSEvent) (interface{}, error) {
		return reactor.OnSNSEvent(ctx, snsEvent)
	}

	lambdaFn, lambdaFnErr := sparta.NewAWSLambda(reactorName(reactor),
		reactorLambda,
		sparta.IAMRoleDefinition{})
	if lambdaFnErr != nil {
		return nil, errors.Wrapf(lambdaFnErr, "attempting to create reactor")
	}

	lambdaFn.Permissions = append(lambdaFn.Permissions, sparta.SNSPermission{
		BasePermission: sparta.BasePermission{
			SourceArn: snsTopic,
		},
	})
	if len(additionalLambdaPermissions) != 0 {
		lambdaFn.RoleDefinition.Privileges = additionalLambdaPermissions
	}
	return lambdaFn, nil
}
1275424
func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config, shouldPreserve func(block *types.Block) bool) (*BlockChain, error) {
	if cacheConfig == nil {
		cacheConfig = &CacheConfig{
			TrieCleanLimit: 256,
			TrieDirtyLimit: 256,
			TrieTimeLimit:  5 * time.Minute,
		}
	}
	bodyCache, _ := lru.New(bodyCacheLimit)
	bodyRLPCache, _ := lru.New(bodyCacheLimit)
	receiptsCache, _ := lru.New(receiptsCacheLimit)
	blockCache, _ := lru.New(blockCacheLimit)
	futureBlocks, _ := lru.New(maxFutureBlocks)
	badBlocks, _ := lru.New(badBlockLimit)

	bc := &BlockChain{
		chainConfig:    chainConfig,
		cacheConfig:    cacheConfig,
		db:             db,
		triegc:         prque.New(nil),
		stateCache:     state.NewDatabaseWithCache(db, cacheConfig.TrieCleanLimit),
		quit:           make(chan struct{}),
		shouldPreserve: shouldPreserve,
		bodyCache:      bodyCache,
		bodyRLPCache:   bodyRLPCache,
		receiptsCache:  receiptsCache,
		blockCache:     blockCache,
		futureBlocks:   futureBlocks,
		engine:         engine,
		vmConfig:       vmConfig,
		badBlocks:      badBlocks,
	}
	bc.validator = NewBlockValidator(chainConfig, bc, engine)
	bc.prefetcher = newStatePrefetcher(chainConfig, bc, engine)
	bc.processor = NewStateProcessor(chainConfig, bc, engine)

	var err error
	bc.hc, err = NewHeaderChain(db, chainConfig, engine, bc.getProcInterrupt)
	if err != nil {
		return nil, err
	}
	bc.genesisBlock = bc.GetBlockByNumber(0)
	if bc.genesisBlock == nil {
		return nil, ErrNoGenesis
	}
	if err := bc.loadLastState(); err != nil {
		return nil, err
	}
	// Check the current state of the block hashes and make sure that we do not have any of the bad blocks in our chain
	for hash := range BadHashes {
		if header := bc.GetHeaderByHash(hash); header != nil {
			// get the canonical block corresponding to the offending header's number
			headerByNumber := bc.GetHeaderByNumber(header.Number.Uint64())
			// make sure the headerByNumber (if present) is in our current canonical chain
			if headerByNumber != nil && headerByNumber.Hash() == header.Hash() {
				log.Error("Found bad hash, rewinding chain", "number", header.Number, "hash", header.ParentHash)
				bc.SetHead(header.Number.Uint64() - 1)
				log.Error("Chain rewind was successful, resuming normal operation")
			}
		}
	}
	// Take ownership of this particular state
	go bc.update()
	return bc, nil
}