func newMiaxMcastServer(c Config, src *miaxMessageSource, i int) (mms *miaxMcastServer) {
	laddr, err := net.ResolveUDPAddr("udp", c.LocalAddr)
	errs.CheckE(err)
	mcaddr, err := net.ResolveUDPAddr("udp", c.FeedAddr)
	errs.CheckE(err)
	laddr.Port += i + LocalPortShift
	mcaddr.Port += i
	mcaddr.IP[net.IPv6len-1] += (byte)(i)
	gapP := c.GapPeriod
	if 0 == gapP {
		gapP = 0xFFFFFFFFFFFFFFFF
	}
	mms = &miaxMcastServer{
		laddr:     laddr,
		mcaddr:    mcaddr,
		src:       src,
		cancel:    make(chan struct{}),
		num:       i,
		gap:       0 != c.GapSize,
		gapSize:   c.GapSize,
		gapPeriod: gapP,
		gapCnt:    0,
	}
	return
}

func (c *conn) ReadMessage() (m Message, err error) {
	defer errs.PassE(&err)
	for c.messageReader.N == 0 {
		errs.CheckE(c.readBsuHeader())
	}

	c.rbuf.Reset()
	var b [2]byte
	var n int
	_, err = io.ReadFull(&c.messageReader, b[:])
	errs.CheckE(err)
	log.Printf("rcv bytes %v", b)
	n, err = c.rbuf.Write(b[:])
	errs.CheckE(err)
	errs.Check(n == len(b))
	h := MessageHeader{
		Length: uint8(b[0]),
		Type:   MessageType(b[1]),
	}
	log.Printf("rcv header %#v", h)
	io.CopyN(&c.rbuf, &c.messageReader, int64(h.Length)-2)
	f := MessageFactory[h.Type]
	errs.Check(f != nil)
	m = f()
	errs.CheckE(binary.Read(&c.rbuf, binary.LittleEndian, m))
	log.Printf("rcv %#v\n", m)
	return
}

func (c *cmdMemh2pcap) ParsingFinished() (err error) {
	if !c.shouldExecute {
		return
	}
	out, err := os.OpenFile(c.OutputFileName, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0644)
	errs.CheckE(err)
	defer out.Close()
	m2p := new(memh2pcap)
	m2p.Open(out)

	for _, fn := range c.inputFileNames {
		fi, err := os.Stat(fn)
		errs.CheckE(err)
		if fi.IsDir() {
			fis, err := ioutil.ReadDir(fn)
			errs.CheckE(err)
			sort.Sort(SortedFiles(fis))
			for _, fi = range fis {
				errs.CheckE(m2p.addFile(filepath.Join(fn, fi.Name())))
			}
		} else {
			errs.CheckE(m2p.addFile(fn))
		}
	}
	return
}

func (c *conn) WriteMessageSimple(m Message) (err error) {
	defer errs.PassE(&err)
	c.pw.SyncStart()
	errs.CheckE(c.pw.WriteMessage(m))
	errs.CheckE(c.pw.Flush())
	return
}

func (m *memh2pcap) addOne(r io.Reader) (err error) {
	defer errs.PassE(&err)
	m.data.Reset()
	scanner := bufio.NewScanner(r)
	for i := 0; scanner.Scan(); i++ {
		v, err := strconv.ParseUint(scanner.Text(), 16, 64)
		errs.CheckE(err)
		line := make([]byte, 8)
		binary.LittleEndian.PutUint64(line, v)
		if i == 0 {
			// skip pseudo header (2 bytes)
			line = line[2:]
		}
		_, err = m.data.Write(line)
		errs.CheckE(err)
	}

	m.sb.Clear()
	b, err := m.sb.AppendBytes(m.data.Len())
	errs.CheckE(err)
	copy(b, m.data.Bytes())
	for _, l := range m.layers {
		errs.CheckE(l.SerializeTo(m.sb, m.so))
	}
	ci := gopacket.CaptureInfo{
		Timestamp:     time.Unix(int64(m.mold.SequenceNumber), 1),
		CaptureLength: len(m.sb.Bytes()),
		Length:        len(m.sb.Bytes()),
	}
	errs.CheckE(m.pw.WritePacket(ci, m.sb.Bytes()))
	m.mold.SequenceNumber++
	return
}

func newBatsFeedMcastServer(c Config, src *batsMessageSource, i int) (fmc *batsFeedMcastServer, err error) {
	laddr, err := net.ResolveUDPAddr("udp", c.LocalAddr)
	errs.CheckE(err)
	mcaddr, err := net.ResolveUDPAddr("udp", c.FeedAddr)
	errs.CheckE(err)
	laddr.Port += i
	mcaddr.Port += i
	mcaddr.IP[net.IPv6len-1] += (byte)(i / 4)
	gapP := int(c.GapPeriod)
	if 0 == gapP {
		gapP = 0xFFFFFFFF
	}
	fmc = &batsFeedMcastServer{
		laddr:     laddr,
		mcaddr:    mcaddr,
		src:       src,
		cancel:    make(chan struct{}),
		num:       i,
		gap:       0 != c.GapSize,
		gapSize:   int(c.GapSize),
		gapPeriod: gapP,
		gapCnt:    0,
	}
	return
}

func (s *batsFeedMcastServer) run() {
	defer s.conn.Close()
	defer s.bmsc.Close()
	ch := s.bmsc.Chan()

	log.Printf("%d ready. source chan %v", s.num, ch)
	for {
		select {
		case _, _ = <-s.cancel:
			log.Printf("%d cancelled", s.num)
			return
		case seq := <-ch:
			if s.gapCheck(seq) {
				log.Printf("%d gap !!! mcast seq %d", s.num, seq)
			} else {
				log.Printf("%d mcast seq %d", s.num, seq)
				m := s.src.GetMessage(seq)
				s.pw.SyncStart()
				errs.CheckE(s.pw.SetSequence(seq))
				errs.CheckE(s.pw.WriteMessage(m))
				errs.CheckE(s.pw.Flush())
			}
		}
	}
}

func createMoldPacket(startSeqNum, count int) (bs []byte, err error) {
	defer errs.PassE(&err)
	type moldUDP64 struct {
		Session        string `struc:"[10]byte"`
		SequenceNumber uint64
		MessageCount   uint16
	}
	type moldUDP64MessageBlock struct {
		MessageLength int16 `struc:"sizeof=Payload"`
		Payload       []uint8
	}

	errs.Check(startSeqNum >= 0)
	errs.Check(count >= 0 && count < 1000)
	mh := moldUDP64{
		Session:        "00TestSess",
		SequenceNumber: uint64(startSeqNum),
		MessageCount:   uint16(count),
	}
	var bb bytes.Buffer
	errs.CheckE(struc.Pack(&bb, &mh))
	for i := 0; i < count; i++ {
		mb := moldUDP64MessageBlock{
			Payload: generateIttoMessage(startSeqNum + i),
		}
		errs.CheckE(struc.Pack(&bb, &mb))
	}
	bs = bb.Bytes()
	return
}

func (s *Subscr) SubscribeFromReader(rd io.Reader) (err error) {
	defer errs.PassE(&err)
	sc := bufio.NewScanner(rd)
	for sc.Scan() {
		text := sc.Text()
		var v uint64
		var b byte
		if _, err := fmt.Sscanf(text, "%c%v", &b, &v); err != nil {
			_, err = fmt.Sscanf(text, "%v", &v)
			errs.CheckE(err)
		}
		oid := packet.OptionIdFromUint64(v)
		if b == 'U' || b == 'u' || b == '!' {
			if s.autoSubscribe {
				s.stoplist[oid] = struct{}{}
			} else {
				s.Unsubscribe(oid)
			}
		} else {
			s.Subscribe(oid)
		}
	}
	errs.CheckE(sc.Err())
	return
}

func NewAvtLogger(w io.Writer, rDict io.Reader) *AvtLogger {
	l := &AvtLogger{
		w:         w,
		TobLogger: *NewTobLogger(),
		stream:    *NewStream(),
	}
	var err error
	l.location, err = time.LoadLocation("EST")
	errs.CheckE(err)
	l.TobLogger.SetUpdateFlags(TobUpdateBothSides)
	if rDict != nil {
		r := csv.NewReader(rDict)
		records, err := r.ReadAll()
		errs.CheckE(err)
		l.oid2AvtName = make(map[int]string)
		for _, rec := range records {
			if len(rec) != 2 {
				log.Fatalf("unexpected dict csv record: %#v\n", rec)
			}
			if rec[0] == "avtSymbol" && rec[1] == "exchangeId" {
				continue // header
			}
			oid, err := strconv.Atoi(rec[1])
			errs.CheckE(err)
			l.oid2AvtName[oid] = rec[0]
		}
	}
	return l
}

func WriteMessage(w io.Writer, m Message) (err error) {
	defer errs.PassE(&err)
	errs.CheckE(m.encodePayload())
	var mt MessageType
	switch m.(type) {
	case *MessageDebug:
		mt = TypeDebug
	case *MessageLoginAccepted:
		mt = TypeLoginAccepted
	case *MessageLoginRejected:
		mt = TypeLoginRejected
	case *MessageSequencedData:
		mt = TypeSequencedData
	case *MessageHeartbeat:
		mt = TypeHeartbeat
	case *MessageEnd:
		mt = TypeEnd
	case *MessageLoginRequest:
		mt = TypeLoginRequest
	case *MessageUnsequencedData:
		mt = TypeUnsequencedData
	case *MessageClientHeartbeat:
		mt = TypeClientHeartbeat
	case *MessageLogout:
		mt = TypeLogout
	}
	errs.CheckE(m.getCommon().setHeader(mt))
	errs.CheckE(binary.Write(w, binary.BigEndian, m.getHeader()))
	n, err := w.Write(m.getCommon().Payload)
	errs.CheckE(err)
	errs.Check(n == len(m.getCommon().Payload))
	return
}

func ReadMessage(r io.Reader) (m Message, err error) {
	defer errs.PassE(&err)
	var mc MessageCommon
	errs.CheckE(binary.Read(r, binary.BigEndian, &mc.Header))
	mc.Payload = make([]byte, mc.Header.Length-1)
	n, err := r.Read(mc.Payload)
	errs.CheckE(err)
	errs.Check(n == len(mc.Payload), n, len(mc.Payload))
	switch mc.Header.Type {
	case TypeDebug:
		m = &MessageDebug{}
	case TypeLoginAccepted:
		m = &MessageLoginAccepted{}
	case TypeLoginRejected:
		m = &MessageLoginRejected{}
	case TypeSequencedData:
		m = &MessageSequencedData{}
	case TypeHeartbeat:
		m = &MessageHeartbeat{}
	case TypeEnd:
		m = &MessageEnd{}
	case TypeLoginRequest:
		m = &MessageLoginRequest{}
	case TypeUnsequencedData:
		m = &MessageUnsequencedData{}
	case TypeClientHeartbeat:
		m = &MessageClientHeartbeat{}
	case TypeLogout:
		m = &MessageLogout{}
	}
	*m.getCommon() = mc
	errs.CheckE(m.decodePayload())
	return
}

func (s *SimLogger) PrintMessage(m efhMessage) (err error) {
	defer errs.PassE(&err)
	var bb bytes.Buffer
	errs.CheckE(binary.Write(&bb, binary.LittleEndian, m))
	if r := bb.Len() % 8; r > 0 {
		// pad to  multiple of 8 bytes
		z := make([]byte, 8)
		_, err = bb.Write(z[0 : 8-r])
		errs.CheckE(err)
	}

	for {
		var qw uint64
		if err := binary.Read(&bb, binary.LittleEndian, &qw); err != nil {
			if err == io.EOF {
				break
			}
			errs.CheckE(err)
		} else {
			s.printfln("DMATOHOST_DATA %016x", qw)
		}
	}
	s.printfln("DMATOHOST_TRAILER 00656e696c616b45")
	return
}

func (c *conn) WriteMessageSimple(m SesMMessage) (err error) {
	defer errs.PassE(&err)
	var b bytes.Buffer
	errs.CheckE(writeSesMHeader(&b, m))
	errs.CheckE(binary.Write(&b, binary.LittleEndian, m))
	_, err = c.rw.Write(b.Bytes())
	return
}

func (m *TomMessageSystemTime) SerializeTo(b gopacket.SerializeBuffer, opts gopacket.SerializeOptions) (err error) {
	defer errs.PassE(&err)
	errs.CheckE(m.TomMessageCommon.SerializeTo(b, opts))
	buf, err := b.AppendBytes(4)
	errs.CheckE(err)
	binary.LittleEndian.PutUint32(buf, m.Second)
	return
}

func (m *IttoMessageBaseReference) SerializeTo(b gopacket.SerializeBuffer, opts gopacket.SerializeOptions) (err error) {
	defer errs.PassE(&err)
	errs.CheckE(m.IttoMessageCommon.SerializeTo(b, opts))
	buf, err := b.AppendBytes(8)
	errs.CheckE(err)
	binary.BigEndian.PutUint64(buf, m.BaseRefNum)
	return
}

func (s *Splitter) SplitByOption(oid packet.OptionId, fileName string) (err error) {
	defer errs.PassE(&err)
	outFile, err := os.Create(fileName)
	errs.CheckE(err)
	defer func() { errs.CheckE(outFile.Close()) }()
	confs := make(map[packet.OptionId]SplitByOptionsConfig)
	confs[oid] = SplitByOptionsConfig{Writer: outFile}
	return s.SplitByOptions(confs)
}

func (c *config) LoadFromReader(rd io.Reader) (err error) {
	defer errs.PassE(&err)
	all, err := ioutil.ReadAll(rd)
	errs.CheckE(err)
	for _, str := range strings.Fields(string(all)) {
		errs.CheckE(c.LoadFromStr(str))
	}
	return
}

func (m *memh2pcap) addFile(filename string) (err error) {
	log.Printf("adding %s\n", filename)
	defer errs.PassE(&err)
	in, err := os.Open(filename)
	errs.CheckE(err)
	errs.CheckE(m.addOne(in))
	errs.CheckE(in.Close())
	return
}

func (c *cmdEfhSuite) genEfhChannels(testDirName string) (cc channels.Config) {
	cc = channels.NewConfig()
	if file, err := os.Open(filepath.Join(testDirName, "channels")); err == nil {
		defer file.Close()
		errs.CheckE(cc.LoadFromReader(file))
	} else {
		errs.CheckE(cc.LoadFromStr(c.Exchange))
	}
	return
}