// Chain returns a Chain that evaluates the given Rules, in order, propagating the (context.Context, *scheduler.Call, error)
// from Rule to Rule. Chain is safe to invoke concurrently.
13885
func (rs Rules) Chain() Chain {
	if len(rs) == 0 {
		return ChainIdentity
	}
	return func(ctx context.Context, e *scheduler.Call, z mesos.Response, err error) (context.Context, *scheduler.Call, mesos.Response, error) {
		return rs[0].Eval(ctx, e, z, err, rs[1:].Chain())
	}
}
423481
func (r Rule) Once() Rule {
	if r == nil {
		return nil
	}
	var once sync.Once
	return func(ctx context.Context, e *scheduler.Event, err error, ch Chain) (context.Context, *scheduler.Event, error) {
		ruleInvoked := false
		once.Do(func() {
			ctx, e, err = r(ctx, e, err, ch)
			ruleInvoked = true
		})
		if !ruleInvoked {
			ctx, e, err = ch(ctx, e, err)
		}
		return ctx, e, err
	}
}
1163819
func (r *ChainRouter) Rule(mo, pattern string, fx FlatHandler) FlatChains {
	if fx == nil {
		fx = IdentityCall
	}
	methods := GetMethods(mo)
	patt := reggy.CreateClassic(pattern)
	fr := FlatRouteBuild(methods, patt, fx, r.Log)
	cr := &ChainRouta{ClassicMatchMux: patt, Handler: fr.Handle}
	r.wg.Lock()
	defer r.wg.Unlock()
	r.paths = append(r.paths, cr)
	return fr
}
602304
func LiftErrors() Rule {
	return func(ctx context.Context, e *scheduler.Event, err error, chain Chain) (context.Context, *scheduler.Event, error) {
		if err != nil {
			return chain(ctx, e, err)
		}
		if e.GetType() == scheduler.Event_ERROR {
			// it's recommended that we abort and re-try subscribing; returning an
			// error here will cause the event loop to terminate and the connection
			// will be reset.
			return chain(ctx, e, ErrEvent(e.GetError().GetMessage()))
		}
		return chain(ctx, e, nil)
	}
}
1348295
func (m Matrix) Chained(next Matrix) Matrix {
	return Matrix{
		next[0]*m[0] + next[2]*m[1],
		next[1]*m[0] + next[3]*m[1],
		next[0]*m[2] + next[2]*m[3],
		next[1]*m[2] + next[3]*m[3],
		next[0]*m[4] + next[2]*m[5] + next[4],
		next[1]*m[4] + next[3]*m[5] + next[5],
	}
}
30421
func (m *Mux) buildChain() {
	m.handler = dispatch{}
	for i := len(m.middleware) - 1; i >= 0; i-- {
		m.handler = m.middleware[i](m.handler)
	}
}
1008982
func (s *Service) loop() {
	// Subscribe to chain events to execute updates on
	var blockchain blockChain
	var txpool txPool
	if s.eth != nil {
		blockchain = s.eth.BlockChain()
		txpool = s.eth.TxPool()
	} else {
		blockchain = s.les.BlockChain()
		txpool = s.les.TxPool()
	}

	chainHeadCh := make(chan core.ChainHeadEvent, chainHeadChanSize)
	headSub := blockchain.SubscribeChainHeadEvent(chainHeadCh)
	defer headSub.Unsubscribe()

	txEventCh := make(chan core.NewTxsEvent, txChanSize)
	txSub := txpool.SubscribeNewTxsEvent(txEventCh)
	defer txSub.Unsubscribe()

	// Start a goroutine that exhausts the subsciptions to avoid events piling up
	var (
		quitCh = make(chan struct{})
		headCh = make(chan *types.Block, 1)
		txCh   = make(chan struct{}, 1)
	)
	go func() {
		var lastTx mclock.AbsTime

	HandleLoop:
		for {
			select {
			// Notify of chain head events, but drop if too frequent
			case head := <-chainHeadCh:
				select {
				case headCh <- head.Block:
				default:
				}

			// Notify of new transaction events, but drop if too frequent
			case <-txEventCh:
				if time.Duration(mclock.Now()-lastTx) < time.Second {
					continue
				}
				lastTx = mclock.Now()

				select {
				case txCh <- struct{}{}:
				default:
				}

			// node stopped
			case <-txSub.Err():
				break HandleLoop
			case <-headSub.Err():
				break HandleLoop
			}
		}
		close(quitCh)
	}()
	// Loop reporting until termination
	for {
		// Resolve the URL, defaulting to TLS, but falling back to none too
		path := fmt.Sprintf("%s/api", s.host)
		urls := []string{path}

		if !strings.Contains(path, "://") { // url.Parse and url.IsAbs is unsuitable (https://github.com/golang/go/issues/19779)
			urls = []string{"wss://" + path, "ws://" + path}
		}
		// Establish a websocket connection to the server on any supported URL
		var (
			conf *websocket.Config
			conn *websocket.Conn
			err  error
		)
		for _, url := range urls {
			if conf, err = websocket.NewConfig(url, "http://localhost/"); err != nil {
				continue
			}
			conf.Dialer = &net.Dialer{Timeout: 5 * time.Second}
			if conn, err = websocket.DialConfig(conf); err == nil {
				break
			}
		}
		if err != nil {
			log.Warn("Stats server unreachable", "err", err)
			time.Sleep(10 * time.Second)
			continue
		}
		// Authenticate the client with the server
		if err = s.login(conn); err != nil {
			log.Warn("Stats login failed", "err", err)
			conn.Close()
			time.Sleep(10 * time.Second)
			continue
		}
		go s.readLoop(conn)

		// Send the initial stats so our node looks decent from the get go
		if err = s.report(conn); err != nil {
			log.Warn("Initial stats report failed", "err", err)
			conn.Close()
			continue
		}
		// Keep sending status updates until the connection breaks
		fullReport := time.NewTicker(15 * time.Second)

		for err == nil {
			select {
			case <-quitCh:
				conn.Close()
				return

			case <-fullReport.C:
				if err = s.report(conn); err != nil {
					log.Warn("Full stats report failed", "err", err)
				}
			case list := <-s.histCh:
				if err = s.reportHistory(conn, list); err != nil {
					log.Warn("Requested history report failed", "err", err)
				}
			case head := <-headCh:
				if err = s.reportBlock(conn, head); err != nil {
					log.Warn("Block stats report failed", "err", err)
				}
				if err = s.reportPending(conn); err != nil {
					log.Warn("Post-block transaction stats report failed", "err", err)
				}
			case <-txCh:
				if err = s.reportPending(conn); err != nil {
					log.Warn("Transaction stats report failed", "err", err)
				}
			}
		}
		// Make sure the connection is closed
		conn.Close()
	}
}
6983
func (api *API) middlewareChain(c APIContexter, w http.ResponseWriter, r *http.Request) {
	for _, middleware := range api.middlewares {
		middleware(c, w, r)
	}
}
472397
func (rs Rules) Eval(ctx context.Context, e *executor.Event, err error, ch Chain) (context.Context, *executor.Event, error) {
	return ch(rs.Chain()(ctx, e, err))
}
1162820
func chainedSingle(rule Rule, next Rule) Rule {
	if next == nil {
		next = Satisfied()
	}

	return &chainedRule{
		Rule: rule,
		next: next,
	}
}
483015
func NewChain(nodes ...Node) *Chain {
	chain := &Chain{}
	for _, node := range nodes {
		chain.nodeGroups = append(chain.nodeGroups, NewNodeGroup(node))
	}
	return chain
}
902233
func GetChain(id string) *Chain {
	for _, c := range chains {
		if c.Name == id {
			return &c
		}
	}

	return nil
}
881563
func (e *Echo) chainMiddleware() {
	if e.head != nil {
		return
	}
	e.head = e.router.Handle(nil)
	for i := len(e.middleware) - 1; i >= 0; i-- {
		e.head = e.ValidMiddleware(e.middleware[i]).Handle(e.head)
	}
}
326240
func commandChain(chain []*cli.Command, args []string) string {
	s := make([]string, 0)
	for _, c := range chain {
		s = append(s, c.Command)
	}
	s = append(s, args...)
	if len(s) > 0 && len(args) == 0 {
		s = append(s, "")
	}
	return strings.Join(s, " ")
}
1272644
func (r *FlatChain) ChainHandleFunc(h http.HandlerFunc) FlatChains {
	fh := FlatChainWrap(h, r.log)
	r.Chain(fh)
	return fh
}
875455
func (b *BatchProvider) Insert(table, chain string, pos int, rulespec ...string) error {

	b.Lock()
	defer b.Unlock()

	if _, ok := b.batchTables[table]; !ok {
		return b.ipt.Insert(table, chain, pos, rulespec...)
	}

	if _, ok := b.rules[table]; !ok {
		b.rules[table] = map[string][]string{}
	}

	if _, ok := b.rules[table][chain]; !ok {
		b.rules[table][chain] = []string{}
	}

	rule := strings.Join(rulespec, " ")

	if pos == 1 {
		b.rules[table][chain] = append([]string{rule}, b.rules[table][chain]...)
	} else if pos > len(b.rules[table][chain]) {
		b.rules[table][chain] = append(b.rules[table][chain], rule)
	} else {
		b.rules[table][chain] = append(b.rules[table][chain], "newvalue")
		copy(b.rules[table][chain][pos-1:], b.rules[table][chain][pos-2:])
		b.rules[table][chain][pos-1] = rule
	}

	return nil
}
952158
func (chain *chainImpl) Halt() {
	select {
	case <-chain.startChan:
		// chain finished starting, so we can halt it
		select {
		case <-chain.haltChan:
			// This construct is useful because it allows Halt() to be called
			// multiple times (by a single thread) w/o panicking. Recal that a
			// receive from a closed channel returns (the zero value) immediately.
			logger.Warningf("[channel: %s] Halting of chain requested again", chain.ChainID())
		default:
			logger.Criticalf("[channel: %s] Halting of chain requested", chain.ChainID())
			// stat shutdown of chain
			close(chain.haltChan)
			// wait for processing of messages to blocks to finish shutting down
			<-chain.doneProcessingMessagesToBlocks
			// close the kafka producer and the consumer
			chain.closeKafkaObjects()
			logger.Debugf("[channel: %s] Closed the haltChan", chain.ChainID())
		}
	default:
		logger.Warningf("[channel: %s] Waiting for chain to finish starting before halting", chain.ChainID())
		<-chain.startChan
		chain.Halt()
	}
}