Add message reader and writer

[alovak]

Context

In some cases, we may need to reply to the incoming message and copy some fields of the request message header into the reply header. Currently, the iso8583-connection package supports message headers (we call it network headers) with the message length only. It also does not link message to the header, so there is no way to use data of the incoming message header in the response.

Summary of the PR

Introduce the message reader and writer interfaces to support implementations that will pack/unpack and write/read messages to/from the network connection letting them use the mechanism of linking headers with requests and responses. We also moved message packing and unpacking right into message writer/reader and write *iso8583.Message into the channels instead of raw data ([]byte).

Changes

• add options to set MessageReader and MessageWriter which if set replace MessageLengthReader and MessageLengthWriter
• unpack message right when we read it from the network and pass *iso8583.Message to the readResponseCh channel instead of passing raw data ([]byte)
• pack message right when we write it to the network, not in the Send
• continue message reading if MessageReader returns nil message and nil error (same as is suggested here)
• rename ErrUnpack into idiomatic UnpackError

If you don't work directly with ErrUnpack, then this is a backward compatible change. but if you do, you should rename the type. I also think we should remove MessageLengthReader and MessageLengthWriter in one of the next releases.

Example of the new use case

Here is an example of how you may use a message reader and writer to link request and response headers:

First, we should create a message reader and writer. They will read/write the message header, read/write the message itself and unpack/pack it. In addition, as shown in the example, you can store request headers in some storage and then use data from the header for the reply:

// messageIO is a helper struct to read/write iso8583 messages from/to
// io.Reader/io.Writer
type messageIO struct {
	Spec           *iso8583.MessageSpec
	requestHeaders map[string]*header
	mu             sync.Mutex
}

func (m *messageIO) ReadMessage(r io.Reader) (*iso8583.Message, error) {
	// read 2 bytes header
	h := &header{}
	err := binary.Read(r, binary.BigEndian, h)
	if err != nil {
		return nil, fmt.Errorf("failed to read message length: %w", err)
	}

	// read message
	rawMessage := make([]byte, h.Length)
	_, err = io.ReadFull(r, rawMessage)
	if err != nil {
		return nil, fmt.Errorf("failed to read message: %w", err)
	}

	// unpack message
	message := iso8583.NewMessage(m.Spec)
	err = message.Unpack(rawMessage)
	if err != nil {
		return nil, fmt.Errorf("failed to unpack message: %w", err)
	}

	mti, err := message.GetMTI()
	if err != nil {
		return nil, fmt.Errorf("failed to get mti: %w", err)
	}

	// if message is request then save header
	if mti[2] == '0' || mti[2] == '2' {
		// save message header to be able to write it back
		// when writing message
		m.mu.Lock()
		defer m.mu.Unlock()

		stan, err := message.GetString(11)
		if err != nil {
			return nil, fmt.Errorf("failed to get stan: %w", err)
		}

		// use stan as a key to save header. it can be stan + rrn
		m.requestHeaders[stan] = h
	}

	return message, nil
}

func (m *messageIO) WriteMessage(w io.Writer, message *iso8583.Message) error {
	// pack message
	rawMessage, err := message.Pack()
	if err != nil {
		return fmt.Errorf("failed to pack message: %w", err)
	}

	// create header with message length
	h := header{
		Length: uint16(len(rawMessage)),
	}

	mti, err := message.GetMTI()
	if err != nil {
		return fmt.Errorf("failed to get mti: %w", err)
	}

	// if message is response then work with saved headers
	// simple check for response mti, use more complex check
	if mti[2] == '1' || mti[2] == '3' {
		// get stan from message
		stan, err := message.GetString(11)
		if err != nil {
			return fmt.Errorf("failed to get stan: %w", err)
		}

		// get header from saved headers
		m.mu.Lock()
		requestHeader, ok := m.requestHeaders[stan]
		m.mu.Unlock()
		if !ok {
			return fmt.Errorf("failed to get header for stan %s", stan)
		}

		h.SourceID = requestHeader.DestID
		h.DestID = requestHeader.SourceID
	}

	// write header
	err = binary.Write(w, binary.BigEndian, h)
	if err != nil {
		return fmt.Errorf("failed to write message length: %w", err)
	}

	// write message
	_, err = w.Write(rawMessage)
	if err != nil {
		return fmt.Errorf("failed to write message: %w", err)
	}

	return nil
}

// header is 2 bytes length of the message
type header struct {
	Length   uint16
	SourceID [4]byte
	DestID   [4]byte
}

Second, set MessageReader and MessageWriter instead of MessageLengthReader and MessageLengthWriter:

	msgIO := &messageIO{Spec: testSpec}

	// we don't need to pass spec and header reader and writer as
	// msgIO will do it all for us
	c, err := connection.New(server.Addr, nil, nil, nil,
		connection.SetMessageReader(msgIO),
		connection.SetMessageWriter(msgIO),
		connection.ErrorHandler(func(err error) {
			require.NoError(t, err)
		}),
	)

Performance Impact

Moving message packing from Send message to the message writer could negatively impact the performance of the package. We have (fixed) and run benchmarks to see that new results a not critically worse than they were before. Here are the new results from running benchmarks on the same machine:

➜ go test -bench=BenchmarkProcess100 -run=XXX -v
goos: darwin
goarch: amd64
pkg: github.com/moov-io/iso8583-connection
cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz
BenchmarkProcess100Messages
BenchmarkProcess100Messages-12               446           2714745 ns/op
BenchmarkProcess1000Messages
BenchmarkProcess1000Messages-12               46          25332634 ns/op
BenchmarkProcess10000Messages
BenchmarkProcess10000Messages-12               4         291991727 ns/op
BenchmarkProcess100000Messages
BenchmarkProcess100000Messages-12              1        2776391366 ns/op

It takes 2.7ms to 100 times: send message (client), receive message (server), respond to the message (server), receive response (client).

[adamdecaf]

This is a nice improvement!

[wadearnold]

So much easier to comprehend!