更新fileaccess服务，添加自动清理文件功能

pkg/queue/queue.go

@@ -4,6 +4,7 @@ package queue
 import (
 	"errors"
 	"sparrow/pkg/serializer"
+
 	"github.com/streadway/amqp"
 )
 

pkg/server/server_manager.go

@@ -4,11 +4,12 @@ package server
 
 import (
 	"errors"
-	"github.com/coreos/etcd/client"
-	"golang.org/x/net/context"
 	"os"
 	"strings"
 	"time"
+
+	"github.com/coreos/etcd/client"
+	"golang.org/x/net/context"
 )
 
 const (
@@ -38,7 +39,7 @@ func NewServerManager(name string, etcd string) (*ServerManager, error) {
 	}, nil
 }
 
-// register server to etcd
+// RegisterServer register server to etcd
 func (mgr *ServerManager) RegisterServer() error {
 	if serverInstance == nil {
 		return errorf(errServerNotInit)
@@ -84,7 +85,7 @@ func (mgr *ServerManager) RegisterServer() error {
 	return nil
 }
 
-// update server hosts
+// UpdateServerHosts update server hosts
 func (mgr *ServerManager) UpdateServerHosts() error {
 	if serverInstance == nil {
 		return errorf(errServerNotInit)
@@ -136,12 +137,12 @@ func (mgr *ServerManager) UpdateServerHosts() error {
 
 	mgr.mapServers = servers
 
-	Log.Infof("UpdateServerHosts is done: %v", mgr.mapServers)
+	//Log.Infof("UpdateServerHosts is done: %v", mgr.mapServers)
 	return nil
 
 }
 
-// get host ips for the server, now return all hosts
+// GetServerHosts get host ips for the server, now return all hosts
 func (mgr *ServerManager) GetServerHosts(serverName string, hostType string) ([]string, error) {
 	server, ok := mgr.mapServers[serverName]
 	if !ok {

services/fileaccess/file.go

@@ -9,11 +9,12 @@ import (
 	"sparrow/pkg/server"
 	"sparrow/pkg/utils"
 	"strconv"
+	"time"
 
 	"github.com/kataras/iris"
 )
 
-func registerRouter(app *iris.Application) {
+func registerRouter(app *iris.Application, fc *FileAccess) {
 	app.Post("/upload_file", iris.LimitRequestBodySize(int64(*conMaxSize)),
 		func(ctx iris.Context) {
 			file, info, err := ctx.FormFile("file")
@@ -58,6 +59,12 @@ func registerRouter(app *iris.Application) {
 			} else {
 				fileURL = "http://" + *conDomain + "/upload/tmp/" + newname
 			}
+			fc.mu.Lock()
+			fc.tempFiles[newname] = &tempFile{
+				fileName:   newfile,
+				createTime: time.Now(),
+			}
+			fc.mu.Unlock()
 			ctx.JSON(map[string]interface{}{
 				"code":    0,
 				"message": "success",

services/fileaccess/fileaccess.go

@@ -10,16 +10,50 @@ import (
 	"sparrow/pkg/server"
 	"sparrow/pkg/utils"
 	"sync"
+	"time"
 )
 
+const checkTimeOut = 30 * time.Minute
+
 // FileAccess RPC服务
 type FileAccess struct {
-	mu sync.RWMutex
+	mu        sync.RWMutex
+	tempFiles map[string]*tempFile
+}
+
+// 代表一个临时文件
+type tempFile struct {
+	//文件路径
+	fileName string
+	//创建时间
+	createTime time.Time
 }
 
 // NewFileAccess create a FileAccessor instance
 func NewFileAccess() *FileAccess {
-	return &FileAccess{}
+	return &FileAccess{
+		tempFiles: make(map[string]*tempFile),
+	}
+}
+
+// TODO:  临时解决文案，下个版本把文件信息写到redis中，利用redis的pub/sub机制，自动清理文件
+func (f *FileAccess) checker() {
+	server.Log.Info("开始文件过期检测")
+	for {
+		for k, v := range f.tempFiles {
+			if time.Now().Sub(v.createTime) > checkTimeOut {
+				//delete file
+				f.mu.Lock()
+				delete(f.tempFiles, k)
+				f.mu.Unlock()
+				err := os.Remove(v.fileName)
+				if err != nil {
+					server.Log.Errorf("自动清理文件失败:%v", err)
+				}
+			}
+		}
+		time.Sleep(30 * time.Minute)
+	}
 }
 
 // MoveFile move a file to new path

services/fileaccess/main.go

@@ -13,6 +13,14 @@ func main() {
 		server.Log.Fatal(err)
 		return
 	}
+
+	fileaccess := NewFileAccess()
+	err = server.RegisterRPCHandler(fileaccess)
+	if err != nil {
+		server.Log.Errorf("RegisterRPCHandler Error: %s", err)
+		return
+	}
+
 	app := iris.New()
 	app.Use(iris.Gzip)
 	iris.WithPostMaxMemory(int64(*conMaxSize))
@@ -25,7 +33,7 @@ func main() {
 	app.Use(cors.New(opts))
 	app.AllowMethods(iris.MethodOptions)
 	app.StaticWeb("/upload", *conStaticPath)
-	registerRouter(app)
+	registerRouter(app, fileaccess)
 	app.Build()
 	// register a http handler
 	err = server.RegisterHTTPHandler(app)
@@ -33,13 +41,9 @@ func main() {
 		server.Log.Errorf("RegisterHTTPHandler Error: %s", err)
 		return
 	}
-	fileaccess := NewFileAccess()
-	err = server.RegisterRPCHandler(fileaccess)
-	if err != nil {
-		server.Log.Errorf("RegisterRPCHandler Error: %s", err)
-		return
-	}
-	// go
+
+	//
+	go fileaccess.checker()
 	err = server.Run()
 	if err != nil {
 		server.Log.Fatal(err)

tests/device/device.go

@@ -157,39 +157,42 @@ func (d *Device) DoLogin() error {
 
 func (d *Device) reportStatus(client *MQTT.Client) {
 
-	payloadHead := protocol.DataHead{
-		Flag:      0,
-		Timestamp: uint64(time.Now().Unix() * 1000),
-	}
-	param := []interface{}{uint8(1)}
-	params, err := tlv.MakeTLVs(param)
-	if err != nil {
-		fmt.Println(err)
-		return
-	}
-	sub := protocol.SubData{
-		Head: protocol.SubDataHead{
-			SubDeviceid: uint16(1),
-			PropertyNum: uint16(1),
-			ParamsCount: uint16(len(params)),
-		},
-		Params: params,
-	}
+	for {
+		time.Sleep(10 * time.Second)
+		payloadHead := protocol.DataHead{
+			Flag:      0,
+			Timestamp: uint64(time.Now().Unix() * 1000),
+		}
+		param := []interface{}{"li jian"}
+		params, err := tlv.MakeTLVs(param)
+		if err != nil {
+			fmt.Println(err)
+			return
+		}
+		sub := protocol.SubData{
+			Head: protocol.SubDataHead{
+				SubDeviceid: uint16(1),
+				PropertyNum: uint16(1),
+				ParamsCount: uint16(len(params)),
+			},
+			Params: params,
+		}
 
-	status := protocol.Data{
-		Head:    payloadHead,
-		SubData: []protocol.SubData{},
-	}
+		status := protocol.Data{
+			Head:    payloadHead,
+			SubData: []protocol.SubData{},
+		}
 
-	status.SubData = append(status.SubData, sub)
+		status.SubData = append(status.SubData, sub)
 
-	payload, err := status.Marshal()
-	if err != nil {
-		fmt.Println(err)
-		return
-	}
+		payload, err := status.Marshal()
+		if err != nil {
+			fmt.Println(err)
+			return
+		}
 
-	client.Publish("s", 1, false, payload)
+		client.Publish("s", 1, false, payload)
+	}
 
 }
 
@@ -297,8 +300,8 @@ func (d *Device) DoAccess() error {
 	}
 
 	// beigin report event test
-	go d.reportEvent(c)
-
+	//go d.reportEvent(c)
+	go d.reportStatus(c)
 	// we just pause here to wait for messages
 	<-make(chan int)