Merge branch 'master' into Fix_warning_and_issues

7 files changed, 91 insertions, 64 deletions

diff --git a/service/http/handler.go b/service/http/handler.go
index 4de33844..3c667035 100644
--- a/service/http/handler.go
+++ b/service/http/handler.go
@@ -2,6 +2,7 @@ package http
 
 import (
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 	"github.com/spiral/roadrunner"
 	"net"
 	"net/http"
@@ -59,6 +60,7 @@ func (e *ResponseEvent) Elapsed() time.Duration {
 // parsed files and query, payload will include parsed form dataTree (if any).
 type Handler struct {
 	cfg *Config
+	log *logrus.Logger
 	rr  *roadrunner.Server
 	mul sync.Mutex
 	lsn func(event int, ctx interface{})
@@ -98,8 +100,8 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// proxy IP resolution
 	h.resolveIP(req)
 
-	req.Open()
-	defer req.Close()
+	req.Open(h.log)
+	defer req.Close(h.log)
 
 	p, err := req.Payload()
 	if err != nil {
diff --git a/service/http/handler_test.go b/service/http/handler_test.go
index 0db999c9..994a663c 100644
--- a/service/http/handler_test.go
+++ b/service/http/handler_test.go
@@ -96,7 +96,7 @@ func TestHandler_Echo(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -197,11 +197,11 @@ func TestHandler_Headers(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
-	time.Sleep(time.Millisecond * 10)
+	time.Sleep(time.Millisecond * 100)
 
 	req, err := http.NewRequest("GET", "http://localhost:8078?hello=world", nil)
 	assert.NoError(t, err)
@@ -260,7 +260,7 @@ func TestHandler_Empty_User_Agent(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -322,7 +322,7 @@ func TestHandler_User_Agent(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -384,7 +384,7 @@ func TestHandler_Cookies(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -451,7 +451,7 @@ func TestHandler_JsonPayload_POST(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -517,7 +517,7 @@ func TestHandler_JsonPayload_PUT(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -534,7 +534,6 @@ func TestHandler_JsonPayload_PUT(t *testing.T) {
 		err := r.Body.Close()
 		if err != nil {
 			t.Errorf("error during the closing Body: error %v", err)
-
 		}
 	}()
 
@@ -579,7 +578,7 @@ func TestHandler_JsonPayload_PATCH(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -641,7 +640,7 @@ func TestHandler_FormData_POST(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -715,7 +714,7 @@ func TestHandler_FormData_POST_Overwrite(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -783,7 +782,7 @@ func TestHandler_FormData_POST_Form_UrlEncoded_Charset(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -857,7 +856,7 @@ func TestHandler_FormData_PUT(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -931,7 +930,7 @@ func TestHandler_FormData_PATCH(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1005,7 +1004,7 @@ func TestHandler_Multipart_POST(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1044,6 +1043,7 @@ func TestHandler_Multipart_POST(t *testing.T) {
 	}
 
 	err = w.WriteField("arr[x][y][e]", "f")
+
 	if err != nil {
 		t.Errorf("error writing the field: error %v", err)
 	}
@@ -1120,7 +1120,7 @@ func TestHandler_Multipart_PUT(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1139,6 +1139,7 @@ func TestHandler_Multipart_PUT(t *testing.T) {
 	}
 
 	err = w.WriteField("name[]", "name1")
+
 	if err != nil {
 		t.Errorf("error writing the field: error %v", err)
 	}
@@ -1235,7 +1236,8 @@ func TestHandler_Multipart_PATCH(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1259,11 +1261,13 @@ func TestHandler_Multipart_PATCH(t *testing.T) {
 	}
 
 	err = w.WriteField("name[]", "name2")
+
 	if err != nil {
 		t.Errorf("error writing the field: error %v", err)
 	}
 
 	err = w.WriteField("name[]", "name3")
+
 	if err != nil {
 		t.Errorf("error writing the field: error %v", err)
 	}
@@ -1350,7 +1354,7 @@ func TestHandler_Error(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1394,7 +1398,7 @@ func TestHandler_Error2(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1438,7 +1442,7 @@ func TestHandler_Error3(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1499,7 +1503,7 @@ func TestHandler_ResponseDuration(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1558,7 +1562,7 @@ func TestHandler_ResponseDurationDelayed(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1617,7 +1621,7 @@ func TestHandler_ErrorDuration(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1689,7 +1693,7 @@ func TestHandler_IP(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1748,7 +1752,7 @@ func TestHandler_XRealIP(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1811,7 +1815,7 @@ func TestHandler_XForwardedFor(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1874,7 +1878,7 @@ func TestHandler_XForwardedFor_NotTrustedRemoteIp(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -1925,7 +1929,7 @@ func BenchmarkHandler_Listen_Echo(b *testing.B) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			b.Errorf("error listening the interface: error %v", err)
 		}
 	}()
diff --git a/service/http/request.go b/service/http/request.go
index 98508342..5d91bfb6 100644
--- a/service/http/request.go
+++ b/service/http/request.go
@@ -3,6 +3,7 @@ package http
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/sirupsen/logrus"
 	"github.com/spiral/roadrunner"
 	"github.com/spiral/roadrunner/service/http/attributes"
 	"io/ioutil"
@@ -112,21 +113,21 @@ func NewRequest(r *http.Request, cfg *UploadsConfig) (req *Request, err error) {
 }
 
 // Open moves all uploaded files to temporary directory so it can be given to php later.
-func (r *Request) Open() {
+func (r *Request) Open(log *logrus.Logger) {
 	if r.Uploads == nil {
 		return
 	}
 
-	r.Uploads.Open()
+	r.Uploads.Open(log)
 }
 
 // Close clears all temp file uploads
-func (r *Request) Close() {
+func (r *Request) Close(log *logrus.Logger) {
 	if r.Uploads == nil {
 		return
 	}
 
-	r.Uploads.Clear()
+	r.Uploads.Clear(log)
 }
 
 // Payload request marshaled RoadRunner payload based on PSR7 data. values encode method is JSON. Make sure to open
diff --git a/service/http/service.go b/service/http/service.go
index 1547538b..abe7b3a7 100644
--- a/service/http/service.go
+++ b/service/http/service.go
@@ -3,6 +3,7 @@ package http
 import (
 	"context"
 	"fmt"
+	"github.com/sirupsen/logrus"
 	"github.com/spiral/roadrunner"
 	"github.com/spiral/roadrunner/service/env"
 	"github.com/spiral/roadrunner/service/http/attributes"
@@ -31,6 +32,8 @@ type middleware func(f http.HandlerFunc) http.HandlerFunc
 // Service manages rr, http servers.
 type Service struct {
 	cfg        *Config
+	log        *logrus.Logger
+	cprod      roadrunner.CommandProducer
 	env        env.Environment
 	lsns       []func(event int, ctx interface{})
 	mdwr       []middleware
@@ -48,6 +51,11 @@ func (s *Service) Attach(w roadrunner.Controller) {
 	s.controller = w
 }
 
+// ProduceCommands changes the default command generator method
+func (s *Service) ProduceCommands(producer roadrunner.CommandProducer) {
+	s.cprod = producer
+}
+
 // AddMiddleware adds new net/http mdwr.
 func (s *Service) AddMiddleware(m middleware) {
 	s.mdwr = append(s.mdwr, m)
@@ -60,8 +68,9 @@ func (s *Service) AddListener(l func(event int, ctx interface{})) {
 
 // Init must return configure svc and return true if svc hasStatus enabled. Must return error in case of
 // misconfiguration. Services must not be used without proper configuration pushed first.
-func (s *Service) Init(cfg *Config, r *rpc.Service, e env.Environment) (bool, error) {
+func (s *Service) Init(cfg *Config, r *rpc.Service, e env.Environment, log *logrus.Logger) (bool, error) {
 	s.cfg = cfg
+	s.log = log
 	s.env = e
 
 	if r != nil {
@@ -87,6 +96,7 @@ func (s *Service) Serve() error {
 		}
 	}
 
+	s.cfg.Workers.CommandProducer = s.cprod
 	s.cfg.Workers.SetEnv("RR_HTTP", "true")
 
 	s.rr = roadrunner.NewServer(s.cfg.Workers)
@@ -132,19 +142,34 @@ func (s *Service) Serve() error {
 
 	if s.http != nil {
 		go func() {
-			err <- s.http.ListenAndServe()
+			httpErr := s.http.ListenAndServe()
+			if httpErr != nil && httpErr != http.ErrServerClosed {
+				err <- httpErr
+			} else {
+				err <- nil
+			}
 		}()
 	}
 
 	if s.https != nil {
 		go func() {
-			err <- s.https.ListenAndServeTLS(s.cfg.SSL.Cert, s.cfg.SSL.Key)
+			httpErr := s.https.ListenAndServeTLS(s.cfg.SSL.Cert, s.cfg.SSL.Key)
+			if httpErr != nil && httpErr != http.ErrServerClosed {
+				err <- httpErr
+			} else {
+				err <- nil
+			}
 		}()
 	}
 
 	if s.fcgi != nil {
 		go func() {
-			err <- s.serveFCGI()
+			httpErr := s.serveFCGI()
+			if httpErr != nil && httpErr != http.ErrServerClosed {
+				err <- httpErr
+			} else {
+				err <- nil
+			}
 		}()
 	}
 
@@ -159,11 +184,10 @@ func (s *Service) Stop() {
 	if s.fcgi != nil {
 		go func() {
 			err := s.fcgi.Shutdown(context.Background())
-			if err != nil {
-				// TODO think about returning error from this Stop function
+			if err != nil && err != http.ErrServerClosed {
 				// Stop() error
 				// push error from goroutines to the channel and block unil error or success shutdown or timeout
-				fmt.Println(fmt.Errorf("error shutting down the server, error: %v", err))
+				s.log.Error(fmt.Errorf("error shutting down the fcgi server, error: %v", err))
 				return
 			}
 		}()
@@ -171,9 +195,9 @@ func (s *Service) Stop() {
 
 	if s.https != nil {
 		go func() {
-			err := s.fcgi.Shutdown(context.Background())
-			if err != nil {
-				fmt.Println(fmt.Errorf("error shutting down the server, error: %v", err))
+			err := s.https.Shutdown(context.Background())
+			if err != nil && err != http.ErrServerClosed {
+				s.log.Error(fmt.Errorf("error shutting down the https server, error: %v", err))
 				return
 			}
 		}()
@@ -181,9 +205,9 @@ func (s *Service) Stop() {
 
 	if s.http != nil {
 		go func() {
-			err := s.fcgi.Shutdown(context.Background())
-			if err != nil {
-				fmt.Println(fmt.Errorf("error shutting down the server, error: %v", err))
+			err := s.http.Shutdown(context.Background())
+			if err != nil && err != http.ErrServerClosed {
+				s.log.Error(fmt.Errorf("error shutting down the http server, error: %v", err))
 				return
 			}
 		}()
diff --git a/service/http/service_test.go b/service/http/service_test.go
index bfc10971..c4b2c2c4 100644
--- a/service/http/service_test.go
+++ b/service/http/service_test.go
@@ -54,9 +54,7 @@ func Test_Service_NoConfig(t *testing.T) {
 	c.Register(ID, &Service{})
 
 	err := c.Init(&testCfg{httpCfg: `{"Enable":true}`})
-	if err != nil {
-		t.Errorf("error during the Init: error %v", err)
-	}
+	assert.Error(t, err)
 
 	s, st := c.Get(ID)
 	assert.NotNil(t, s)
diff --git a/service/http/uploads.go b/service/http/uploads.go
index 7efa7e4a..8a46f230 100644
--- a/service/http/uploads.go
+++ b/service/http/uploads.go
@@ -3,6 +3,7 @@ package http
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/sirupsen/logrus"
 	"io"
 	"io/ioutil"
 	"mime/multipart"
@@ -46,16 +47,15 @@ func (u *Uploads) MarshalJSON() ([]byte, error) {
 
 // Open moves all uploaded files to temp directory, return error in case of issue with temp directory. File errors
 // will be handled individually.
-func (u *Uploads) Open() {
+func (u *Uploads) Open(log *logrus.Logger) {
 	var wg sync.WaitGroup
 	for _, f := range u.list {
 		wg.Add(1)
 		go func(f *FileUpload) {
 			defer wg.Done()
 			err := f.Open(u.cfg)
-			if err != nil {
-				// TODO handle error mechanism in goroutines
-				fmt.Println(fmt.Errorf("error opening the file: error %v", err))
+			if err != nil && log != nil {
+				log.Error(fmt.Errorf("error opening the file: error %v", err))
 			}
 		}(f)
 	}
@@ -64,13 +64,12 @@ func (u *Uploads) Open() {
 }
 
 // Clear deletes all temporary files.
-func (u *Uploads) Clear() {
+func (u *Uploads) Clear(log *logrus.Logger) {
 	for _, f := range u.list {
 		if f.TempFilename != "" && exists(f.TempFilename) {
 			err := os.Remove(f.TempFilename)
-			if err != nil {
-				// TODO error handling mechanism
-				fmt.Println(fmt.Errorf("error removing the file: error %v", err))
+			if err != nil && log != nil {
+				log.Error(fmt.Errorf("error removing the file: error %v", err))
 			}
 		}
 	}
@@ -131,7 +130,6 @@ func (f *FileUpload) Open(cfg *UploadsConfig) (err error) {
 		err = file.Close()
 	}()
 
-
 	tmp, err := ioutil.TempFile(cfg.TmpDir(), "upload")
 	if err != nil {
 		// most likely cause of this issue is missing tmp dir
diff --git a/service/http/uploads_test.go b/service/http/uploads_test.go
index c5de224b..1890c02b 100644
--- a/service/http/uploads_test.go
+++ b/service/http/uploads_test.go
@@ -51,7 +51,7 @@ func TestHandler_Upload_File(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -138,7 +138,7 @@ func TestHandler_Upload_NestedFile(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -225,7 +225,7 @@ func TestHandler_Upload_File_NoTmpDir(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()
@@ -312,7 +312,7 @@ func TestHandler_Upload_File_Forbids(t *testing.T) {
 
 	go func() {
 		err := hs.ListenAndServe()
-		if err != nil {
+		if err != nil && err != http.ErrServerClosed {
 			t.Errorf("error listening the interface: error %v", err)
 		}
 	}()