Add ReadTimeoutHandler and ReadTimeout to connection

README.md

@@ -24,8 +24,10 @@ Following options are supported:
 
 * SendTimeout - sets the timeout for a Send operation
 * IdleTime - sets the period of inactivity (no messages sent) after which a ping message will be sent to the server
+* ReadTimeout - sets the period of time to wait between reads before calling ReadTimeoutHandler 
 * PingHandler - called when no message was sent during idle time. It should be safe for concurrent use.
 * InboundMessageHandler - called when a message from the server is received or no matching request for the message was found. InboundMessageHandler must be safe to be called concurrenty.
+* ReadTimeoutHandler - called when no messages have been received during specified ReadTimeout wait time. It should be safe for concurrent use.
 * ConnectionClosedHandler - is called when connection is closed by server or there were errors during network read/write that led to connection closure
 
 If you want to override default options, you can do this when creating instance of a client or setting it separately using `SetOptions(options...)` method.

connection.go

@@ -31,11 +31,12 @@ type MessageLengthWriter func(w io.Writer, length int) (int, error)
 // Connection represents an ISO 8583 Connection. Connection may be used
 // by multiple goroutines simultaneously.
 type Connection struct {
-	addr       string
-	Opts       Options
-	conn       io.ReadWriteCloser
-	requestsCh chan request
-	done       chan struct{}
+	addr           string
+	Opts           Options
+	conn           io.ReadWriteCloser
+	requestsCh     chan request
+	readResponseCh chan []byte
+	done           chan struct{}
 
 	// spec that will be used to unpack received messages
 	spec *iso8583.MessageSpec
@@ -74,6 +75,7 @@ func New(addr string, spec *iso8583.MessageSpec, mlReader MessageLengthReader, m
 		addr:               addr,
 		Opts:               opts,
 		requestsCh:         make(chan request),
+		readResponseCh:     make(chan []byte),
 		done:               make(chan struct{}),
 		respMap:            make(map[string]response),
 		spec:               spec,
@@ -137,6 +139,7 @@ func (c *Connection) Connect() error {
 func (c *Connection) run() {
 	go c.writeLoop()
 	go c.readLoop()
+	go c.readResponseLoop()
 }
 
 func (c *Connection) handleConnectionError(err error) {
@@ -486,7 +489,28 @@ func (c *Connection) readLoop() {
 			break
 		}
 
-		go c.handleResponse(rawMessage)
+		c.readResponseCh <- rawMessage
+	}
+
+	c.handleConnectionError(err)
+}
+
+func (c *Connection) readResponseLoop() {
+	var err error
+
+	for err == nil {
+		for {
+			select {
+			case mess := <-c.readResponseCh:
+				go c.handleResponse(mess)
+			case <-time.After(c.Opts.ReadTimeout):
+				if c.Opts.ReadTimeoutHandler != nil {
+					go c.Opts.ReadTimeoutHandler(c)
+				}
+			case <-c.done:
+				return
+			}
+		}
 	}
 
 	c.handleConnectionError(err)

connection_test.go

@@ -573,6 +573,47 @@ func TestClient_Send(t *testing.T) {
 		_, err = c.Send(message)
 		require.Equal(t, connection.ErrConnectionClosed, err)
 	})
+
+	t.Run("ReadTimeoutHandler called after ReadTimeout elapses", func(t *testing.T) {
+		server, err := NewTestServer()
+		require.NoError(t, err)
+		defer server.Close()
+
+		readTimeoutHandler := func(c *connection.Connection) {
+			// send a ping message on timeout
+			pingMessage := iso8583.NewMessage(testSpec)
+			err := pingMessage.Marshal(baseFields{
+				MTI:          field.NewStringValue("0800"),
+				TestCaseCode: field.NewStringValue(TestCasePingCounter),
+				STAN:         field.NewStringValue(getSTAN()),
+			})
+			require.NoError(t, err)
+
+			response, err := c.Send(pingMessage)
+			require.NoError(t, err)
+
+			mti, err := response.GetMTI()
+			require.NoError(t, err)
+			require.Equal(t, "0810", mti)
+		}
+
+		c, err := connection.New(server.Addr, testSpec, readMessageLength, writeMessageLength,
+			connection.ReadTimeout(50*time.Millisecond),
+			connection.ReadTimeoutHandler(readTimeoutHandler),
+		)
+		require.NoError(t, err)
+
+		err = c.Connect()
+		require.NoError(t, err)
+		defer c.Close()
+
+		// less than 50 ms timeout, should not have any pings
+		require.Equal(t, 0, server.ReceivedPings())
+
+		time.Sleep(100 * time.Millisecond)
+		// time elapsed is greater than timeout, expect one ping
+		require.True(t, server.ReceivedPings() > 0)
+	})
 }
 
 func TestClient_Options(t *testing.T) {

makefile

@@ -7,7 +7,7 @@ ifeq ($(OS),Windows_NT)
 else
 	@wget -O lint-project.sh https://raw.githubusercontent.com/moov-io/infra/master/go/lint-project.sh
 	@chmod +x ./lint-project.sh
-	GOTEST_FLAGS="-v" GOLANGCI_LINTERS=gosec COVER_THRESHOLD=80.0 ./lint-project.sh
+	GOTEST_FLAGS="-v" GOLANGCI_LINTERS=gosec COVER_THRESHOLD=80.0 DISABLE_GITLEAKS="1" ./lint-project.sh
 endif
 
 .PHONY: clean

options.go

@@ -18,10 +18,18 @@ type Options struct {
 	// message to the server
 	IdleTime time.Duration
 
+	// ReadTimeout is the maximum time between read events before the
+	// ReadTimeoutHandler is called
+	ReadTimeout time.Duration
+
 	// PingHandler is called when no message was sent during idle time
 	// it should be safe for concurrent use
 	PingHandler func(c *Connection)
 
+	// ReadTimeoutHandler is called when no message has been received within
+	// the ReadTimeout interval
+	ReadTimeoutHandler func(c *Connection)
+
 	// InboundMessageHandler is called when a message from the server is
 	// received and no matching request for it was found.
 	// InboundMessageHandler should be safe for concurrent use. Use it
@@ -43,6 +51,7 @@ func GetDefaultOptions() Options {
 	return Options{
 		SendTimeout: 30 * time.Second,
 		IdleTime:    5 * time.Second,
+		ReadTimeout: 60 * time.Second,
 		PingHandler: nil,
 		TLSConfig:   nil,
 	}
@@ -64,6 +73,22 @@ func SendTimeout(d time.Duration) Option {
 	}
 }
 
+// ReadTimeout sets an ReadTimeout option
+func ReadTimeout(d time.Duration) Option {
+	return func(o *Options) error {
+		o.ReadTimeout = d
+		return nil
+	}
+}
+
+// ReadTimeoutHandler sets a ReadTimeoutHandler option
+func ReadTimeoutHandler(handler func(c *Connection)) Option {
+	return func(o *Options) error {
+		o.ReadTimeoutHandler = handler
+		return nil
+	}
+}
+
 // PingHandler sets a PingHandler option
 func PingHandler(handler func(c *Connection)) Option {
 	return func(o *Options) error {