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 (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 (s *replayServer) run() {
	type moldudp64request struct {
		Session        string
		SequenceNumber uint64
		MessageCount   uint16
	}

	laddr, err := net.ResolveUDPAddr("udp", s.laddr)
	errs.CheckE(err)
	conn, err := net.ListenUDP("udp", laddr)
	errs.CheckE(err)
	defer conn.Close()
	buf := make([]byte, 20, 65536)
	for {
		n, addr, err := conn.ReadFromUDP(buf)
		errs.CheckE(err)
		if n != 20 {
			log.Printf("ignore wrong request from %s: %v\n", addr, buf)
			continue
		}
		req := &moldudp64request{
			Session:        string(buf[0:10]),
			SequenceNumber: binary.BigEndian.Uint64(buf[10:18]),
			MessageCount:   binary.BigEndian.Uint16(buf[18:20]),
		}
		go func() {
			const MAX_MESSAGES = (1500 - 34 - 20) / 40
			log.Printf("got request: %v\n", req)
			//num := int(req.SequenceNumber) - 10
			num := int(req.MessageCount)
			if num <= 0 {
				num = 1
			} else if num > MAX_MESSAGES {
				num = MAX_MESSAGES
			}
			resp, err := createMoldPacket(int(req.SequenceNumber), num)
			errs.CheckE(err)
			errs.Check(len(resp) < 1500-34)

			if s.sleepEnabled {
				sleep := time.Duration(250+2500/num) * time.Millisecond
				log.Printf("sleeping for %s\n", sleep)
				time.Sleep(sleep)
			}
			log.Printf("send response: %v\n", resp)
			n, err = conn.WriteToUDP(resp, addr)
			errs.CheckE(err)
			errs.Check(n == len(resp), n, len(resp))
		}()
	}
}

func (p *reusingProcessor) ProcessPacket(data []byte, ci gopacket.CaptureInfo, decoded []gopacket.DecodingLayer) (err error) {
	defer errs.PassE(&err)
	p.pkt = reusingPacket{
		data:   data,
		ci:     ci,
		layers: decoded,
	}
	p.m = applicationMessage{
		flows:     p.m.flows[:0],
		timestamp: p.pkt.Timestamp(),
	}
	p.handler.HandlePacket(&p.pkt)
	for _, layer := range decoded {
		//log.Printf("%#v", layer)
		switch l := layer.(type) {
		case gopacket.NetworkLayer:
			p.m.flows = append(p.m.flows, l.NetworkFlow())
		case *layers.UDP:
			p.m.flows = append(p.m.flows, l.TransportFlow())
		case *miax.MachTop:
			p.messageNum = l.MachPackets
			p.messageIndex = 0
			p.nextSeqNums = p.nextSeqNums[:0]
		case *miax.Mach:
			errs.Check(len(p.nextSeqNums) < p.messageNum, len(p.nextSeqNums), p.messageNum)
			p.nextSeqNums = append(p.nextSeqNums, l.SequenceNumber)
		case miax.TomMessage:
			errs.Check(p.messageIndex < p.messageNum, p.messageIndex, p.messageNum)
			p.m.layer = l
			p.m.seqNum = p.nextSeqNums[p.messageIndex]
			p.messageIndex++
			p.handler.HandleMessage(&p.m)
		case *bats.BSU:
			p.m.seqNum = uint64(l.Sequence)
		case bats.PitchMessage:
			p.m.layer = l
			p.handler.HandleMessage(&p.m)
			p.m.seqNum++
		case *nasdaq.MoldUDP64:
			p.m.seqNum = l.SequenceNumber
		case nasdaq.IttoMessage:
			p.m.layer = l
			p.handler.HandleMessage(&p.m)
			p.m.seqNum++
		}
	}
	return
}

func parseSymbol(data []byte) packet.OptionId {
	errs.Check(len(data) >= 6)
	var b [8]byte
	copy(b[:], data)
	oid := packet.OptionIdFromUint64(binary.LittleEndian.Uint64(b[:]))
	return oid
}

func (mc *MessageCommon) setHeader(Type MessageType) (err error) {
	defer errs.PassE(&err)
	mc.Header.Type = Type
	mc.Header.Length = uint8(MessageLength[Type])
	errs.Check(mc.Header.Length != 0)
	return
}

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 *SesMServerConn) login() (err error) {
	defer errs.PassE(&err)
	m, err := s.mconn.ReadMessage() //////// miax conn
	errs.CheckE(err)
	lr, ok := m.(*miax.SesMLoginRequest)
	errs.Check(ok)
	// проверка, что мы не хотим рефреш
	errs.Check(0 == lr.ReqSeqNum)
	res := miax.SesMLoginResponse{
		LoginStatus:   miax.LoginStatusSuccess,
		HighestSeqNum: uint64(s.src.CurrentSequence()),
	}
	errs.CheckE(s.mconn.WriteMessageSimple(&res))
	log.Printf("login done")
	return
}

func (m *Mach) DecodeFromBytes(data []byte, df gopacket.DecodeFeedback) (err error) {
	defer errs.PassE(&err)
	errs.Check(len(data) >= 12)
	length := binary.LittleEndian.Uint16(data[8:10])
	errs.Check(len(data) >= int(length))
	*m = Mach{
		SequenceNumber: binary.LittleEndian.Uint64(data[0:8]),
		Length:         length,
		Type:           data[10],
		SessionNumber:  data[11],
		BaseLayer: layers.BaseLayer{
			Contents: data[:12],
			Payload:  data[12:length],
		},
	}
	return nil
}

func (p *packetWriter) WriteMessage(m Message) (err error) {
	defer errs.PassE(&err)
	m.getCommon().setHeader(m.Type())
	errs.CheckE(binary.Write(&p.mb, binary.LittleEndian, m))
	p.bh.Count++
	errs.Check(p.bh.Count != 0)
	return
}

func (c *conn) readSize() (err error) {
	defer errs.PassE(&err)
	errs.Check(c.messageReader.N == 0, c.messageReader.N)
	var s uint16
	errs.CheckE(binary.Read(c.rw, binary.LittleEndian, &s))
	log.Printf("rcv SesM size %d\n", s)
	c.messageReader = io.LimitedReader{R: c.rw, N: int64(s)}
	return
}

func (mc *MessageCommon) setHeader(Type MessageType) error {
	var payloadSize int
	switch Type {
	case TypeDebug, TypeSequencedData, TypeUnsequencedData:
		payloadSize = len(mc.Payload)
	case TypeLoginAccepted:
		payloadSize = 30
	case TypeLoginRejected:
		payloadSize = 1
	case TypeLoginRequest:
		payloadSize = 46
	}
	errs.Check(payloadSize == len(mc.Payload), payloadSize, len(mc.Payload))
	errs.Check(payloadSize < math.MaxUint16)
	mc.Header.Type = Type
	mc.Header.Length = uint16(payloadSize) + 1
	return nil
}

func (m *MessageLoginAccepted) encodePayload() (err error) {
	m.Payload = make([]byte, 30)
	copy(m.Payload[0:10], []byte(m.Session))
	errs.Check(m.SequenceNumber > 0)
	copy(m.Payload[10:30], []byte(fmt.Sprintf("%020d", m.SequenceNumber)))
	m.Header.Type = TypeLoginAccepted
	m.Header.Length = uint16(len(m.Payload) + 1)
	return nil
}

func (f *tomLayerFactory) Create(layerType gopacket.LayerType) gopacket.DecodingLayer {
	d := int(layerType - gopacket.LayerType(TOM_LAYERS_BASE_NUM))
	if d < 0 || d > 255 {
		panic("FIXME")
		//return gopacket.LayerTypeZero // FIXME
	}
	m := TomMessageTypeMetadata[d]
	errs.Check(m.LayerType == layerType)
	return m.CreateLayer()
}

func (b *book) ApplyOperation(operation SimOperation) {
	oid := operation.GetOptionId()
	errs.Check(oid.Valid())
	os, ok := b.options[oid]
	if !ok {
		os = NewOptionState(b.newOptionSideState)
		b.options[oid] = os
	}
	os.Side(operation.GetSide()).updateLevel(operation)
}

func (s *spinServerConn) run() {
	defer errs.Catch(func(ce errs.CheckerError) {
		log.Printf("caught %s\n", ce)
	})
	defer s.conn.Close()
	errs.CheckE(s.login())
	cancelSendImageAvail := make(chan struct{})
	defer func() {
		// close channel only if not already closed
		select {
		case _, ok := <-cancelSendImageAvail:
			if !ok {
				return
			}
		default:
		}
		close(cancelSendImageAvail)
	}()
	go s.sendImageAvail(cancelSendImageAvail)

	m, err := s.bconn.ReadMessage()
	errs.CheckE(err)
	req, ok := m.(*bats.MessageSpinRequest)
	errs.Check(ok)
	close(cancelSendImageAvail)

	seq := s.src.CurrentSequence()
	errs.Check(int(req.Sequence) <= seq, req.Sequence, seq)
	res := bats.MessageSpinResponse{
		Sequence: req.Sequence,
		Count:    uint32(seq) - req.Sequence + 1,
		Status:   bats.SpinStatusAccepted,
	}
	errs.CheckE(s.bconn.WriteMessageSimple(&res))
	errs.CheckE(s.sendAll(int(req.Sequence), seq+1))
	s.waitForMcast(seq)
	res2 := bats.MessageSpinFinished{
		Sequence: req.Sequence,
	}
	errs.CheckE(s.bconn.WriteMessageSimple(&res2))
	log.Println("spin finished")
}

func NewNasdaqExchangeSimulatorServer(c Config) (es ExchangeSimulator, err error) {
	errs.Check(c.Protocol == "nasdaq")
	errs.Check(!c.Interactive)
	es = &exchangeNasdaq{
		glimpse: &glimpseServer{
			laddr:          ":16001",
			snapshotSeqNum: 5,
		},
		replay: &replayServer{
			laddr:        ":17001",
			sleepEnabled: true,
		},
		mcast: &mcastServer{
			laddr: c.LocalAddr,
			raddr: c.FeedAddr,
			seq:   1000,
			pps:   1,
		},
	}
	return
}

func (p *packetWriter) Flush() (err error) {
	defer errs.PassE(&err)
	length := p.mb.Len() + 8
	errs.Check(length <= math.MaxUint16, length)
	p.bh.Length = uint16(length)
	errs.CheckE(binary.Write(&p.pb, binary.LittleEndian, p.bh))
	_, err = p.mb.WriteTo(&p.pb)
	errs.CheckE(err)
	_, err = p.w.Write(p.pb.Bytes())
	errs.CheckE(err)
	p.Reset()
	return
}

func (c *conn) readBsuHeader() (err error) {
	defer errs.PassE(&err)
	errs.Check(c.messageReader.N == 0, c.messageReader.N)
	var h BsuHeader
	errs.CheckE(binary.Read(c.rw, binary.LittleEndian, &h))
	log.Printf("rcv BSU header %#v", h)
	if h.Sequence != 0 {
		errs.Check(h.Sequence == c.nextSeqNum, h.Sequence, c.nextSeqNum)
		c.nextSeqNum += uint32(h.Count)
	}
	if h.Unit == 0 {
		errs.Check(h.Sequence == 0, h.Sequence)
	} else {
		if c.unit == 0 {
			c.unit = int(h.Unit)
		} else {
			errs.Check(c.unit == int(h.Unit), c.unit, h.Unit)
		}
	}
	c.messageReader = io.LimitedReader{R: c.rw, N: int64(h.Length) - 8}
	return
}