Merge pull request #38 from getsumio/develop

Develop

Author: getsumio

internal/servers/onpremise.go

@@ -8,6 +8,7 @@ import (
 	"path"
 	"regexp"
 	"sync"
+	"time"
 
 	"github.com/getsumio/getsum/internal/config"
 	. "github.com/getsumio/getsum/internal/config"
@@ -32,7 +33,6 @@ const uuidPattern = "^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-4[a-fA-F0-9]{3}-[8|9|aA|bB][
 var regex *regexp.Regexp = regexp.MustCompile(uuidPattern)
 
 const default_capacity = 250
-const threshold = 150
 
 //start server in given config listen address and port or tls details
 //TODO add interface support
@@ -102,7 +102,6 @@ func handlePost(s *OnPremiseServer, w http.ResponseWriter, r *http.Request) {
 	var algorithm = ValueOf(&config.Algorithm[0])
 	config.Dir = &s.StoragePath
 	supplier, err := factory.GetSupplierByAlgo(config, &algorithm)
-	//something went  wrong, TODO add error handler
 	if err != nil {
 		handleError("Can not create algorithm runner instance: "+err.Error(), w)
 		return
@@ -117,6 +116,7 @@ func handlePost(s *OnPremiseServer, w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	w.Write(jsonStat)
+	supplier.Data().StartTime = time.Now()
 	go supplier.Run()
 	s.suppliers[processId] = supplier
 	logger.Info("Process started")
@@ -185,15 +185,12 @@ func handleError(message string, w http.ResponseWriter, params ...interface{}) {
 
 func (s *OnPremiseServer) ensureCapacity() {
 	if len(s.suppliers) >= default_capacity {
-		var i int = 0
+		now := time.Now()
 		for k := range s.suppliers {
-			if i >= threshold {
-				supplier := s.suppliers[k]
-				if supplier.Status().Type >= status.RUNNING {
-					delete(s.suppliers, k)
-				}
+			supplier := s.suppliers[k]
+			if int(now.Sub(supplier.Data().StartTime).Seconds()) > (supplier.Data().TimeOut * 2) {
+				delete(s.suppliers, k)
 			}
-			i++
 		}
 	}
 }

internal/supplier/gosupplier.go

@@ -66,6 +66,7 @@ func (s *GoSupplier) Run() {
 	s.status.Type = status.STARTED
 	t := time.After(time.Duration(s.TimeOut) * time.Second)
 	stat := make(chan string)
+	errChan := make(chan string)
 	//collect related Hash instance
 	hash, err := getHash(s.Algorithm, s.Key)
 	if err != nil {
@@ -74,7 +75,7 @@ func (s *GoSupplier) Run() {
 	}
 
 	//start calculation
-	go calculate(hash, stat, s.File)
+	go calculate(hash, stat, errChan, s.File)
 	for {
 		tEnd := time.Now()
 		took := tEnd.Sub(tStart)
@@ -91,6 +92,10 @@ func (s *GoSupplier) Run() {
 			s.status.Value = fmt.Sprintf("%dms", took.Milliseconds())
 			s.status.Checksum = strings.Fields(val)[0]
 			return
+		case val := <-errChan:
+			s.status.Type = status.ERROR
+			s.status.Value = val
+			return
 		default:
 			//still running update time
 			s.status.Type = status.RUNNING
@@ -112,7 +117,6 @@ func (s *GoSupplier) Delete() {
 }
 
 //terminate
-//TODO figure out how to terminate hash.sum() function
 func (s *GoSupplier) Terminate() error {
 	if s.status.Type == status.RUNNING {
 		s.status.Type = status.TERMINATED
@@ -126,10 +130,12 @@ func (s *GoSupplier) Data() *BaseSupplier {
 }
 
 //calculates given file and returns checksum
-func calculate(hash hash.Hash, status chan string, file *file.File) {
+func calculate(hash hash.Hash, status chan string, errChan chan string, file *file.File) {
 	f, _ := os.Open(file.Path())
 	defer f.Close()
 	if _, err := io.Copy(hash, f); err != nil {
+		errChan <- err.Error()
+		return
 	}
 	status <- hex.EncodeToString(hash.Sum(nil))
 }

internal/supplier/supplier.go

@@ -1,6 +1,8 @@
 package supplier
 
 import (
+	"time"
+
 	"github.com/getsumio/getsum/internal/file"
 	"github.com/getsumio/getsum/internal/status"
 )
@@ -22,4 +24,5 @@ type BaseSupplier struct {
 	File      *file.File
 	TimeOut   int
 	Key       string
+	StartTime time.Time
 }