Method handler

receiver/replayer.go:112–146

* * Users should modify this function according to different demands. */

()

 * Users should modify this function according to different demands.
 */
func (er *ExampleReplayer) handler() {
	for msg := range er.pendingQueue {
		count := uint64(len(msg.message.RawLogs))
		if count == 0 {
			// probe request
			continue
		}

		// parse batched message
		oplogs := make([]oplog.ParsedLog, len(msg.message.RawLogs))
		for i, raw := range msg.message.RawLogs {
			oplogs[i] = oplog.ParsedLog{}
			if err := bson.Unmarshal(raw, &oplogs[i]); err != nil {
				// impossible switch, need panic and exit
				l.Logger.Panicf("unmarshal oplog[%v] failed[%v]", raw, err)
				return
			}
			l.Logger.Infof("%v", oplogs[i]) // just print for test, users can modify to fulfill different needs
			// fmt.Println(oplogs[i])
		}

		if callback := msg.completion; callback != nil {
			callback() // exec callback
		}

		// get the newest timestamp
		n := len(oplogs)
		lastTs := utils.TimeStampToInt64(oplogs[n-1].Timestamp)
		er.Ack = lastTs

		l.Logger.Debugf("handle ack[%v]", er.Ack)

		// add logical code below
	}
}