BCDA-446: Download data issue - truncated file when requesting from A…

…WS (#65)

* Separate API and data file router and servers

* Panic on Serve() error

* Configurable http.Server timeouts

bcda/main.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/http"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -119,11 +120,25 @@ func setUpApp() *cli.App {
 					autoMigrate()
 				}
 
-				err = http.ListenAndServe(":3000", NewRouter())
-				if err != nil {
-					return err
+				api := &http.Server{
+					Handler:      NewAPIRouter(),
+					ReadTimeout:  time.Duration(getEnvInt("API_READ_TIMEOUT", 10)) * time.Second,
+					WriteTimeout: time.Duration(getEnvInt("API_WRITE_TIMEOUT", 20)) * time.Second,
+					IdleTimeout:  time.Duration(getEnvInt("API_IDLE_TIMEOUT", 120)) * time.Second,
 				}
 
+				fileserver := &http.Server{
+					Handler:      NewDataRouter(),
+					ReadTimeout:  time.Duration(getEnvInt("FILESERVER_READ_TIMEOUT", 10)) * time.Second,
+					WriteTimeout: time.Duration(getEnvInt("FILESERVER_WRITE_TIMEOUT", 360)) * time.Second,
+					IdleTimeout:  time.Duration(getEnvInt("FILESERVER_IDLE_TIMEOUT", 120)) * time.Second,
+				}
+
+				smux := NewServiceMux(":3000")
+				smux.AddServer(fileserver, "/data")
+				smux.AddServer(api, "")
+				smux.Serve()
+
 				return nil
 			},
 		},
@@ -353,3 +368,14 @@ func createAlphaToken() (string, error) {
 	aco, user, tokenString, err := authBackend.CreateAlphaToken()
 	return fmt.Sprintf("%s\n%s\n%s", aco.Name, user.Name, tokenString), err
 }
+
+func getEnvInt(varName string, defaultVal int) int {
+	v := os.Getenv(varName)
+	if v != "" {
+		i, err := strconv.Atoi(v)
+		if err == nil {
+			return i
+		}
+	}
+	return defaultVal
+}

bcda/router.go

@@ -12,8 +12,7 @@ import (
 	"github.com/go-chi/chi"
 )
 
-//NewRouter provides a router with all the required... routes
-func NewRouter() http.Handler {
+func NewAPIRouter() http.Handler {
 	r := chi.NewRouter()
 	r.Use(auth.ParseToken, logging.NewStructuredLogger())
 	// Serve up the swagger ui folder
@@ -34,11 +33,16 @@ func NewRouter() http.Handler {
 			r.Get("/bb_metadata", blueButtonMetadata)
 		}
 	})
-	r.With(auth.RequireTokenAuth, auth.RequireTokenACOMatch).Get("/data/{jobID}/{acoID}.ndjson", serveData)
 	r.Get("/_version", getVersion)
 	return r
 }
 
+func NewDataRouter() http.Handler {
+	r := chi.NewRouter()
+	r.With(auth.ParseToken, logging.NewStructuredLogger(), auth.RequireTokenAuth, auth.RequireTokenACOMatch).Get("/data/{jobID}/{acoID}.ndjson", serveData)
+	return r
+}
+
 // FileServer conveniently sets up a http.FileServer handler to serve
 // static files from a http.FileSystem.
 // stolen from https://github.com/go-chi/chi/blob/master/_examples/fileserver/main.go

bcda/router_test.go

@@ -2,35 +2,39 @@ package main
 
 import (
 	"fmt"
-	"github.com/go-chi/chi"
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
 	"os"
 	"testing"
 
+	"github.com/go-chi/chi"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/suite"
 )
 
 type RouterTestSuite struct {
 	suite.Suite
-	server *httptest.Server
-	rr     *httptest.ResponseRecorder
+	apiServer  *httptest.Server
+	dataServer *httptest.Server
+	rr         *httptest.ResponseRecorder
 }
 
 func (suite *RouterTestSuite) SetupTest() {
 	os.Setenv("DEBUG", "true")
-	suite.server = httptest.NewServer(NewRouter())
+	suite.apiServer = httptest.NewServer(NewAPIRouter())
+	suite.dataServer = httptest.NewServer(NewDataRouter())
 	suite.rr = httptest.NewRecorder()
 }
 
 func (suite *RouterTestSuite) TearDownTest() {
-	suite.server.Close()
+	suite.apiServer.Close()
+	suite.dataServer.Close()
 }
 
 func (suite *RouterTestSuite) GetStringBody(url string) (string, error) {
-	client := suite.server.Client()
+	client := suite.apiServer.Client()
 	res, err := client.Get(url)
 	if err != nil {
 		return "", err
@@ -45,18 +49,18 @@ func (suite *RouterTestSuite) GetStringBody(url string) (string, error) {
 }
 
 func (suite *RouterTestSuite) TestDefaultRoute() {
-	r, err := suite.GetStringBody(suite.server.URL)
+	r, err := suite.GetStringBody(suite.apiServer.URL)
 	assert.Nil(suite.T(), err, fmt.Sprintf("Error when getting default route: %s", err))
 	assert.Equal(suite.T(), "Hello world!", r, "Default route returned wrong body")
 }
 
 func (suite *RouterTestSuite) TestDataRoute() {
-	_, err := suite.GetStringBody(suite.server.URL + "/data")
+	_, err := suite.GetStringBody(suite.dataServer.URL + "/data")
 	assert.Nil(suite.T(), err, fmt.Sprintf("Error when getting data route: %s", err))
 	//assert.Equal(suite.T(), "Hello world!", r, "Default route returned wrong body")
 }
 func (suite *RouterTestSuite) TestFileServerRoute() {
-	_, err := suite.GetStringBody(suite.server.URL + "/api/v1/swagger")
+	_, err := suite.GetStringBody(suite.apiServer.URL + "/api/v1/swagger")
 	assert.Nil(suite.T(), err, fmt.Sprintf("Error when getting swagger route: %s", err))
 	r := chi.NewRouter()
 	// Set up a bad route.  DON'T do this in real life

bcda/servicemux.go

@@ -0,0 +1,110 @@
+package main
+
+import (
+	"bufio"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/soheilhy/cmux"
+)
+
+var keepAliveInterval int = 30
+
+func init() {
+	interval := os.Getenv("SERVICE_MUX_KEEP_ALIVE_INTERVAL")
+	if interval != "" {
+		interval, err := strconv.Atoi(interval)
+		if err != nil {
+			panic(err)
+		}
+		keepAliveInterval = interval
+	}
+}
+
+type tcpKeepAliveListener struct {
+	*net.TCPListener
+}
+
+func (ln tcpKeepAliveListener) Accept() (net.Conn, error) {
+	tc, err := ln.AcceptTCP()
+	if err != nil {
+		return nil, err
+	}
+
+	err = tc.SetKeepAlive(true)
+	if err != nil {
+		return nil, err
+	}
+
+	err = tc.SetKeepAlivePeriod(time.Duration(keepAliveInterval) * time.Second)
+	if err != nil {
+		return nil, err
+	}
+
+	return tc, nil
+}
+
+func URLPrefixMatcher(prefix string) cmux.Matcher {
+	return func(r io.Reader) bool {
+		req, err := http.ReadRequest(bufio.NewReader(r))
+		if err != nil {
+			return false
+		}
+		return strings.HasPrefix(req.URL.Path, prefix)
+	}
+}
+
+type ServiceMux struct {
+	Addr     string
+	Listener net.Listener
+	Servers  []map[*http.Server]string
+}
+
+func NewServiceMux(addr string) *ServiceMux {
+	s := &ServiceMux{
+		Addr: addr,
+	}
+
+	ln, err := net.Listen("tcp", s.Addr)
+	if err != nil {
+		panic(err)
+	}
+
+	s.Listener = ln
+	return s
+}
+
+func (sm *ServiceMux) AddServer(s *http.Server, m string) {
+	var server = make(map[*http.Server]string)
+	server[s] = m
+	sm.Servers = append(sm.Servers, server)
+}
+
+func (sm *ServiceMux) Serve() {
+	m := cmux.New(tcpKeepAliveListener{sm.Listener.(*net.TCPListener)})
+
+	for _, server := range sm.Servers {
+		for srv, path := range server {
+			var match net.Listener
+
+			if path == "" {
+				match = m.Match(cmux.Any())
+			} else {
+				match = m.Match(URLPrefixMatcher(path))
+			}
+
+			//nolint
+			go srv.Serve(match)
+		}
+	}
+
+	err := m.Serve()
+	if err != nil {
+		panic(err)
+	}
+}