// Replay msgs in file or start the console
550148
func (cs *ConsensusState) ReplayFile(file string, console bool) error {

	if cs.IsRunning() {
		return errors.New("cs is already running, cannot replay")
	}
	if cs.wal != nil {
		return errors.New("cs wal is open, cannot replay")
	}

	cs.startForReplay()

	// ensure all new step events are regenerated as expected

	ctx := context.Background()
	newStepSub, err := cs.eventBus.Subscribe(ctx, subscriber, types.EventQueryNewRoundStep)
	if err != nil {
		return errors.Errorf("failed to subscribe %s to %v", subscriber, types.EventQueryNewRoundStep)
	}
	defer cs.eventBus.Unsubscribe(ctx, subscriber, types.EventQueryNewRoundStep)

	// just open the file for reading, no need to use wal
	fp, err := os.OpenFile(file, os.O_RDONLY, 0600)
	if err != nil {
		return err
	}

	pb := newPlayback(file, fp, cs, cs.state.Copy())
	defer pb.fp.Close() // nolint: errcheck

	var nextN int // apply N msgs in a row
	var msg *TimedWALMessage
	for {
		if nextN == 0 && console {
			nextN = pb.replayConsoleLoop()
		}

		msg, err = pb.dec.Decode()
		if err == io.EOF {
			return nil
		} else if err != nil {
			return err
		}

		if err := pb.cs.readReplayMessage(msg, newStepSub); err != nil {
			return err
		}

		if nextN > 0 {
			nextN--
		}
		pb.count++
	}
}
744817
func (bl *Logger) startLogger() {
	for {
		select {
		case bm := <-bl.msg:
			for _, l := range bl.outputs {
				err := l.WriteMsg(bm.msg, bm.level)
				if err != nil {
					fmt.Println("ERROR, unable to WriteMsg:", err)
				}
			}
		}
	}
}
198771
func (c *Connection) msgCallback(msgID uint16, data []byte) {
	if c == nil {
		log.Warn("Already disconnected, ignoring the message.")
		return
	}

	msg, ok := c.msgMap[msgID]
	if !ok {
		log.Warnf("Unknown message received, ID: %d", msgID)
		return
	}

	// decode message context to fix for special cases of messages,
	// for example:
	// - replies that don't have context as first field (comes as zero)
	// - events that don't have context at all (comes as non zero)
	//
	context, err := c.codec.DecodeMsgContext(data, msg)
	if err != nil {
		log.Errorf("decoding context failed: %v", err)
	}

	chanID, isMulti, seqNum := unpackRequestContext(context)
	if log.Level == logger.DebugLevel { // for performance reasons - logrus does some processing even if debugs are disabled
		log.WithFields(logger.Fields{
			"context":  context,
			"msg_id":   msgID,
			"msg_size": len(data),
			"channel":  chanID,
			"is_multi": isMulti,
			"seq_num":  seqNum,
			"msg_crc":  msg.GetCrcString(),
		}).Debugf("<-- govpp recv: %s", msg.GetMessageName())
	}

	if context == 0 || c.isNotificationMessage(msgID) {
		// process the message as a notification
		c.sendNotifications(msgID, data)
		return
	}

	// match ch according to the context
	c.channelsLock.RLock()
	ch, ok := c.channels[chanID]
	c.channelsLock.RUnlock()
	if !ok {
		log.WithFields(logger.Fields{
			"channel": chanID,
			"msg_id":  msgID,
		}).Error("Channel ID not known, ignoring the message.")
		return
	}

	// if this is a control ping reply to a multipart request,
	// treat this as a last part of the reply
	lastReplyReceived := isMulti && msgID == c.pingReplyID

	// send the data to the channel, it needs to be copied,
	// because it will be freed after this function returns
	sendReply(ch, &vppReply{
		msgID:        msgID,
		seqNum:       seqNum,
		data:         append([]byte(nil), data...),
		lastReceived: lastReplyReceived,
	})

	// store actual time of this reply
	c.lastReplyLock.Lock()
	c.lastReply = time.Now()
	c.lastReplyLock.Unlock()
}
787236
func (pb *playback) replayReset(count int, newStepSub types.Subscription) error {
	pb.cs.Stop()
	pb.cs.Wait()

	newCS := NewConsensusState(pb.cs.config, pb.genesisState.Copy(), pb.cs.blockExec,
		pb.cs.blockStore, pb.cs.txNotifier, pb.cs.evpool)
	newCS.SetEventBus(pb.cs.eventBus)
	newCS.startForReplay()

	if err := pb.fp.Close(); err != nil {
		return err
	}
	fp, err := os.OpenFile(pb.fileName, os.O_RDONLY, 0600)
	if err != nil {
		return err
	}
	pb.fp = fp
	pb.dec = NewWALDecoder(fp)
	count = pb.count - count
	fmt.Printf("Reseting from %d to %d\n", pb.count, count)
	pb.count = 0
	pb.cs = newCS
	var msg *TimedWALMessage
	for i := 0; i < count; i++ {
		msg, err = pb.dec.Decode()
		if err == io.EOF {
			return nil
		} else if err != nil {
			return err
		}
		if err := pb.cs.readReplayMessage(msg, newStepSub); err != nil {
			return err
		}
		pb.count++
	}
	return nil
}
167954
func (c *Network) ReplayXHRWithParams(v *NetworkReplayXHRParams) (*gcdmessage.ChromeResponse, error) {
	return gcdmessage.SendDefaultRequest(c.target, c.target.GetSendCh(), &gcdmessage.ParamRequest{Id: c.target.GetId(), Method: "Network.replayXHR", Params: v})
}
309669
func replayEvents(store Store, doneCh <-chan struct{}) <-chan string {
	var names []string
	eventKeyCh := make(chan string)

	go func() {
		retryTimer := time.NewTimer(retryInterval)
		defer retryTimer.Stop()
		defer close(eventKeyCh)
		for {
			names = store.ListN(100)
			for _, name := range names {
				select {
				case eventKeyCh <- strings.TrimSuffix(name, eventExt):
					// Get next key.
				case <-doneCh:
					return
				}
			}

			if len(names) < 2 {
				retryTimer.Reset(retryInterval)
				select {
				case <-retryTimer.C:
				case <-doneCh:
					return
				}
			}
		}
	}()

	return eventKeyCh
}
1225093
func (rl *MemoryReplayLog) Start() error {
	rl.batches = make(map[string]*ReplaySet)
	rl.entries = make(map[HashPrefix]uint32)
	return nil
}
80623
func Fail(msg string) {
	logger := blog.Get()
	logger.AuditErr(msg)
	fmt.Fprintf(os.Stderr, msg)
	os.Exit(1)
}
665609
func readWriteMsg(c msgio.ReadWriter, out []byte) ([]byte, error) {
	wresult := make(chan error)
	go func() {
		wresult <- c.WriteMsg(out)
	}()

	msg, err1 := c.ReadMsg()

	// Always wait for the read to finish.
	err2 := <-wresult

	if err1 != nil {
		return nil, err1
	}
	if err2 != nil {
		c.ReleaseMsg(msg)
		return nil, err2
	}
	return msg, nil
}
1101217
func (l *intLogger) Trace(msg string, args ...interface{}) {
	l.Log(Trace, msg, args...)
}
806225
func PgXlogReplayPause(db XODB) error {
	var err error

	// sql query
	const sqlstr = `SELECT pg_catalog.pg_xlog_replay_pause()`

	// run query
	XOLog(sqlstr)
	_, err = db.Exec(sqlstr)
	return err
}
509472
func (s *ScrubWriter) WriteMsg(m *dns.Msg) error {
	state := Request{Req: s.req, W: s.ResponseWriter}

	n := state.Scrub(m)
	state.SizeAndDo(n)
	return s.ResponseWriter.WriteMsg(n)
}
918568
func (kvmsg *Kvmsg) Dump() {
	size := kvmsg.Size()
	body, _ := kvmsg.GetBody()
	seq, _ := kvmsg.GetSequence()
	key, _ := kvmsg.GetKey()
	fmt.Fprintf(os.Stderr, "[seq:%v][key:%v][size:%v]", seq, key, size)
	for char_nbr := 0; char_nbr < size; char_nbr++ {
		fmt.Fprintf(os.Stderr, "%02X", body[char_nbr])
	}
	fmt.Fprintln(os.Stderr)
}
1060330
func (msg Msg) Discard() error {
	_, err := io.Copy(ioutil.Discard, msg.Payload)
	return err
}
2685
func (l *Logger) TraceMsg(message string, keyvals ...interface{}) error {
	return Msg(l.Trace, message, keyvals...)
}
84853
func main() {
	logger.Level = golog.INFO

	app.Version("1.0.2")

	// install dependencies
	outputCmd("go", []string{"get", "github.com/ivpusic/rerun"})

	switch kingpin.MustParse(app.Parse(os.Args[1:])) {
	case run.FullCommand():
		handleRunCommand()
	case newCmd.FullCommand():
		handleNewCommand()
	default:
		app.Usage(os.Stdout)
	}
}
277077
func (ds *DummyStatus) PrintMsg(msg message) {
	if !globalJSON {
		console.Println(msg.String())
	} else {
		console.Println(msg.JSON())
	}
}