提交httpservice与example示例

This commit is contained in:
duanhf2012
2020-04-07 19:33:27 +08:00
parent bedb715c6d
commit f2eda1c635
10 changed files with 438 additions and 460 deletions

View File

@@ -42,20 +42,21 @@ func (slf *Cluster) ReadAllSubNetConfig() error {
clusterCfgPath :=strings.TrimRight(configdir,"/") +"/cluster" clusterCfgPath :=strings.TrimRight(configdir,"/") +"/cluster"
fileInfoList,err := ioutil.ReadDir(clusterCfgPath) fileInfoList,err := ioutil.ReadDir(clusterCfgPath)
if err != nil { if err != nil {
return err return fmt.Errorf("Read dir %s is fail :%+v",clusterCfgPath,err)
} }
slf.mapSubNetInfo =map[string] SubNet{} slf.mapSubNetInfo =map[string] SubNet{}
for _,f := range fileInfoList{ for _,f := range fileInfoList{
if f.IsDir() == true { if f.IsDir() == true {
subnetinfo,err:=slf.ReadClusterConfig(strings.TrimRight(strings.TrimRight(clusterCfgPath,"/"),"\\")+"/"+f.Name()+"/"+"cluster.json") filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath,"/"),"\\")+"/"+f.Name()+"/"+"cluster.json"
subnetinfo,err:=slf.ReadClusterConfig(filePath)
if err != nil { if err != nil {
return err return fmt.Errorf("read file path %s is error:%+v" ,filePath,err)
} }
slf.mapSubNetInfo[f.Name()] = *subnetinfo slf.mapSubNetInfo[f.Name()] = *subnetinfo
} }
} }
return nil return nil
} }
@@ -63,14 +64,16 @@ func (slf *Cluster) ReadLocalSubNetServiceConfig(subnet string) error {
clusterCfgPath :=strings.TrimRight(configdir,"/") +"/cluster" clusterCfgPath :=strings.TrimRight(configdir,"/") +"/cluster"
fileInfoList,err := ioutil.ReadDir(clusterCfgPath) fileInfoList,err := ioutil.ReadDir(clusterCfgPath)
if err != nil { if err != nil {
return err return fmt.Errorf("Read %s dir is fail:%+v ",clusterCfgPath,err)
} }
slf.mapSubNetInfo =map[string] SubNet{} slf.mapSubNetInfo =map[string] SubNet{}
for _,f := range fileInfoList{ for _,f := range fileInfoList{
if f.IsDir() == true && f.Name()==subnet{ //同一子网 if f.IsDir() == true && f.Name()==subnet{ //同一子网
localNodeServiceCfg,err:=slf.ReadServiceConfig(strings.TrimRight(strings.TrimRight(clusterCfgPath,"/"),"\\")+"/"+f.Name()+"/"+"service.json") filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath,"/"),"\\")+"/"+f.Name()+"/"+"service.json"
localNodeServiceCfg,err:=slf.ReadServiceConfig(filePath)
if err != nil { if err != nil {
return err return fmt.Errorf("Read file %s is fail :%+v",filePath,err)
} }
slf.localNodeServiceCfg =localNodeServiceCfg slf.localNodeServiceCfg =localNodeServiceCfg
} }
@@ -89,8 +92,10 @@ func (slf *Cluster) InitCfg(currentNodeId int) error{
localNodeMapService := map[string]interface{}{} //本Node支持的服务 localNodeMapService := map[string]interface{}{} //本Node支持的服务
localNodeInfo := NodeInfo{} localNodeInfo := NodeInfo{}
err := slf.ReadAllSubNetConfig() err := slf.ReadAllSubNetConfig()
if err != nil {
return err
}
//分析配置 //分析配置
var localSubnetName string var localSubnetName string
@@ -160,8 +165,7 @@ func (slf *Cluster) InitCfg(currentNodeId int) error{
slf.localNodeInfo =localNodeInfo slf.localNodeInfo =localNodeInfo
//读取服务 //读取服务
slf.ReadLocalSubNetServiceConfig(slf.localsubnet.SubNetName) return slf.ReadLocalSubNetServiceConfig(slf.localsubnet.SubNetName)
return err
} }

View File

@@ -81,6 +81,9 @@ func (slf *EventProcessor) GetEventReciver() IEventProcessor{
return slf.eventReciver return slf.eventReciver
} }
type IHttpEventData interface {
Handle()
}
func (slf *EventProcessor) EventHandler(processor IEventProcessor,ev *Event) { func (slf *EventProcessor) EventHandler(processor IEventProcessor,ev *Event) {
defer func() { defer func() {
@@ -92,5 +95,21 @@ func (slf *EventProcessor) EventHandler(processor IEventProcessor,ev *Event) {
} }
}() }()
if slf.innerEventHandler(ev) == true {
return
}
processor.OnEventHandler(ev) processor.OnEventHandler(ev)
} }
func (slf *EventProcessor) innerEventHandler(ev *Event) bool {
switch ev.Type {
case Sys_Event_Http_Event:
ev.Data.(IHttpEventData).Handle()
return true
}
return false
}

View File

@@ -13,15 +13,30 @@ import (
type GateService struct { type GateService struct {
service.Service service.Service
processor network.Processor processor network.Processor
httpRouter sysservice.IHttpRouter
} }
func (slf *GateService) OnInit() error{ func (slf *GateService) OnInit() error{
tcpervice := node.GetService("TcpService").(*sysservice.TcpService) tcpervice := node.GetService("TcpService").(*sysservice.TcpService)
slf.processor = &processor.PBProcessor{} slf.processor = &processor.PBProcessor{}
tcpervice.SetProcessor(slf.processor) tcpervice.SetProcessor(slf.processor)
httpervice := node.GetService("HttpService").(*sysservice.HttpService)
slf.httpRouter = sysservice.NewHttpHttpRouter(slf)
httpervice.SetHttpRouter(slf.httpRouter)
slf.httpRouter.RegRouter(sysservice.METHOD_GET,"/get/query",slf.HttpTest)
slf.httpRouter.SetServeFile(sysservice.METHOD_GET,"/img/head/","d:/img")
return nil return nil
} }
func (slf *GateService) HttpTest(session *sysservice.HttpSession) {
session.SetHeader("a","b")
session.Write([]byte("this is a test"))
v,_:=session.Query("a")
v2,_:=session.Query("b")
fmt.Print(string(session.GetBody()),"\n",v,"\n",v2)
}
func (slf *GateService) OnEventHandler(ev *event.Event) error{ func (slf *GateService) OnEventHandler(ev *event.Event) error{
if ev.Type == event.Sys_Event_Tcp_RecvPack { if ev.Type == event.Sys_Event_Tcp_RecvPack {

View File

@@ -5,7 +5,7 @@
"ListenAddr":"127.0.0.1:8001", "ListenAddr":"127.0.0.1:8001",
"NodeName": "Node_Test1", "NodeName": "Node_Test1",
"remark":"//以_打头的表示只在本机进程不对整个子网开发", "remark":"//以_打头的表示只在本机进程不对整个子网开发",
"ServiceList": ["TestService1","TestService2","TestServiceCall","GateService","TcpService"] "ServiceList": ["TestService1","TestService2","TestServiceCall","GateService","TcpService","HttpService"]
} }
] ]
} }

View File

@@ -1,6 +1,16 @@
{ {
"HttpService":{ "HttpService":{
"Port":10001 "ListenAddr":"0.0.0.0:9402",
"ReadTimeout":10000,
"WriteTimeout":10000,
"ProcessTimeout":10000,
"CAFile":[
{
"Certfile":"",
"Keyfile":""
}
]
}, },
"TcpService":{ "TcpService":{
"ListenAddr":"0.0.0.0:9030", "ListenAddr":"0.0.0.0:9030",

View File

@@ -236,7 +236,10 @@ func main(){
tcpService := &sysservice.TcpService{} tcpService := &sysservice.TcpService{}
gateService := &GateService.GateService{} gateService := &GateService.GateService{}
tcpService.SetEventReciver(gateService) tcpService.SetEventReciver(gateService)
node.Setup(tcpService,gateService)
httpService := &sysservice.HttpService{}
node.Setup(tcpService,gateService,httpService)
node.OpenProfilerReport(time.Second*10) node.OpenProfilerReport(time.Second*10)
node.Start() node.Start()
} }

View File

@@ -77,6 +77,6 @@ func (slf *HttpServer) startListen() error {
} }
func (slf *HttpServer) SetHttps(caFile []CAFile) { func (slf *HttpServer) SetCAFile(caFile []CAFile) {
slf.caFileList = caFile slf.caFileList = caFile
} }

View File

@@ -68,22 +68,26 @@ func initNode(id int){
//1.初始化集群 //1.初始化集群
err := cluster.GetCluster().Init(GetNodeId()) err := cluster.GetCluster().Init(GetNodeId())
if err != nil { if err != nil {
panic(err) log.Fatal("read system config is error %+v",err)
} }
//2.service模块初始化 //2.service模块初始化
service.Init(closeSig) service.Init(closeSig)
//3.初始化预安装的服务 //3.setup service
for _,s := range preSetupService { for _,s := range preSetupService {
pServiceCfg := cluster.GetCluster().GetServiceCfg(s.GetName())
s.Init(s,cluster.GetRpcClient,cluster.GetRpcServer,pServiceCfg)
//是否配置的service //是否配置的service
if cluster.GetCluster().IsConfigService(s.GetName()) == false { if cluster.GetCluster().IsConfigService(s.GetName()) == false {
continue continue
} }
service.Setup(s) service.Setup(s)
} }
//4.init service
for _,s := range preSetupService {
pServiceCfg := cluster.GetCluster().GetServiceCfg(s.GetName())
s.Init(s,cluster.GetRpcClient,cluster.GetRpcServer,pServiceCfg)
}
} }
func Start() { func Start() {

View File

@@ -195,3 +195,4 @@ func (slf *Service) GetProfiler() *profiler.Profiler{
return slf.profiler return slf.profiler
} }

View File

@@ -1,49 +1,31 @@
package sysservice package sysservice
import ( import (
"encoding/json"
"errors"
"fmt" "fmt"
"io" "github.com/duanhf2012/origin/event"
"io/ioutil"
"net/http"
"os"
"reflect"
"runtime"
"strings"
"time"
"github.com/duanhf2012/origin/sysmodule"
"github.com/duanhf2012/origin/rpc"
"github.com/duanhf2012/origin/cluster"
"github.com/duanhf2012/origin/network" "github.com/duanhf2012/origin/network"
"github.com/duanhf2012/origin/service" "github.com/duanhf2012/origin/service"
"github.com/duanhf2012/origin/util/uuid" "github.com/duanhf2012/origin/util/uuid"
"io"
"io/ioutil"
"net/http"
"os"
"strings"
"time"
) )
var Default_ReadTimeout time.Duration = time.Second*10
var Default_WriteTimeout time.Duration = time.Second*10
var Default_ProcessTimeout time.Duration = time.Second*10
var Default_HttpRouter *HttpRouter= &HttpRouter{}
//http redirect //http redirect
type HttpRedirectData struct { type HttpRedirectData struct {
Url string Url string
CookieList []*http.Cookie CookieList []*http.Cookie
} }
type HttpRequest struct {
URL string
Header http.Header
Body string
paramStr string //http://127.0.0.1:7001/aaa/bbb?aa=1 paramStr is:aa=1
mapParam map[string]string
}
type HttpRespone struct {
Respone []byte
RedirectData HttpRedirectData
}
type HTTP_METHOD int type HTTP_METHOD int
const ( const (
@@ -52,67 +34,95 @@ const (
METHOD_POST METHOD_POST
METHOD_INVALID METHOD_INVALID
//METHOD_PUT HTTP_METHOD = 2
) )
type HttpHandle func(session *HttpSession)
type ControllerMapsType map[string]reflect.Value
type RouterType int
const (
FuncCall RouterType = iota+1
StaticResource
)
type routerMatchData struct { type routerMatchData struct {
callpath string
matchURL string matchURL string
routerType int8 //0表示函数调用 1表示静态资源 httpHandle HttpHandle
} }
type serveHTTPRouterMux struct {
httpFiltrateList [] HttpFiltrate type routerServeFileData struct {
allowOrigin bool matchUrl string
localPath string
method HTTP_METHOD
} }
type IHttpRouter interface {
RegRouter(method HTTP_METHOD, url string, handle HttpHandle) bool
Router(session *HttpSession)
PutHttpSession(httpSession *HttpSession)
SetServeFile(method HTTP_METHOD, urlpath string, dirname string) error
SetFormFileKey(formFileKey string)
GetFormFileKey()string
}
type HttpRouter struct {
pathRouter map[HTTP_METHOD] map[string] routerMatchData //url地址对应本service地址
serveFileData map[string] *routerServeFileData
eventReciver event.IEventProcessor
httpFiltrateList [] HttpFiltrate
formFileKey string
}
type HttpSession struct {
httpRouter IHttpRouter
r *http.Request
w http.ResponseWriter
//parse result
mapParam map[string]string
body []byte
//processor result
statusCode int
msg []byte
fileData *routerServeFileData
redirectData *HttpRedirectData
sessionDone chan *HttpSession
}
type HttpService struct { type HttpService struct {
service.Service service.Service
httpServer network.HttpServer httpServer network.HttpServer
controllerMaps ControllerMapsType
serverHTTPMux serveHTTPRouterMux
postAliasUrl map[HTTP_METHOD] map[string]routerMatchData //url地址对应本service地址 postAliasUrl map[HTTP_METHOD] map[string]routerMatchData //url地址对应本service地址
staticRouterResource map[HTTP_METHOD] routerStaticResoutceData httpRouter IHttpRouter
listenAddr string
httpProcessor IHttpProcessor allowOrigin bool
} processTimeout time.Duration
type routerStaticResoutceData struct {
localPath string
method HTTP_METHOD
} }
type HttpHandle func(request *HttpRequest, resp *HttpRespone) error
func AnalysisRouterUrl(url string) (string, error) { func NewHttpHttpRouter(eventReciver event.IEventProcessor) IHttpRouter {
httpRouter := &HttpRouter{}
//替换所有空格 httpRouter.eventReciver = eventReciver
url = strings.ReplaceAll(url, " ", "") httpRouter.pathRouter =map[HTTP_METHOD] map[string] routerMatchData{}
if len(url) <= 1 || url[0] != '/' { httpRouter.serveFileData = map[string] *routerServeFileData{}
return "", fmt.Errorf("url %s format is error!", url) httpRouter.formFileKey = "file"
for i:=METHOD_NONE+1;i<METHOD_INVALID;i++{
httpRouter.pathRouter[i] = map[string] routerMatchData{}
} }
//去掉尾部的/
return strings.Trim(url, "/"), nil return httpRouter
} }
func (slf *HttpRequest) Query(key string) (string, bool) {
func (slf *HttpSession) Query(key string) (string, bool) {
if slf.mapParam == nil { if slf.mapParam == nil {
slf.mapParam = make(map[string]string) slf.mapParam = make(map[string]string)
//分析字符串
slf.paramStr = strings.Trim(slf.paramStr, "/") paramStr := strings.Trim(slf.r.URL.RawQuery, "/")
paramStrList := strings.Split(slf.paramStr, "&") paramStrList := strings.Split(paramStr, "&")
for _, val := range paramStrList { for _, val := range paramStrList {
index := strings.Index(val, "=") index := strings.Index(val, "=")
if index >= 0 { if index >= 0 {
@@ -125,124 +135,249 @@ func (slf *HttpRequest) Query(key string) (string, bool) {
return ret, ok return ret, ok
} }
type HttpSession struct { func (slf *HttpSession) GetBody() []byte{
request *HttpRequest return slf.body
resp *HttpRespone
} }
type HttpProcessor struct { func (slf *HttpSession) GetMethod() HTTP_METHOD {
httpSessionChan chan *HttpSession return slf.getMethod(slf.r.Method)
pathRouter map[HTTP_METHOD] map[string] routerMatchData //url地址对应本service地址
staticRouterResource map[HTTP_METHOD] routerStaticResoutceData
} }
var Default_HttpSessionChannelNum = 100000 func (slf *HttpSession) GetPath() string{
return strings.Trim(slf.r.URL.Path,"/")
func NewHttpProcessor() *HttpProcessor{
httpProcessor := &HttpProcessor{}
httpProcessor.httpSessionChan =make(chan *HttpSession,Default_HttpSessionChannelNum)
httpProcessor.staticRouterResource = map[HTTP_METHOD] routerStaticResoutceData{}
httpProcessor.pathRouter =map[HTTP_METHOD] map[string] routerMatchData{}
for i:=METHOD_NONE+1;i<METHOD_INVALID;i++{
httpProcessor.pathRouter[i] = map[string] routerMatchData{}
} }
return httpProcessor func (slf *HttpSession) SetHeader(key, value string) {
slf.w.Header().Set(key,value)
} }
var Default_HttpProcessor *HttpProcessor= NewHttpProcessor() func (slf *HttpSession) AddHeader(key, value string) {
slf.w.Header().Add(key,value)
type IHttpProcessor interface {
PutHttpSession(httpSession *HttpSession)
GetHttpSessionChannel() chan *HttpSession
RegRouter(method HTTP_METHOD, url string, handle HttpHandle)
Router(session *HttpSession)
} }
func (slf *HttpProcessor) PutHttpSession(httpSession *HttpSession){ func (slf *HttpSession) WriteStatusCode(statusCode int){
slf.httpSessionChan <- httpSession slf.statusCode = statusCode
} }
func (slf *HttpProcessor) GetHttpSessionChannel() chan *HttpSession { func (slf *HttpSession) Write(msg []byte) {
return slf.httpSessionChan slf.msg = msg
} }
func (slf *HttpProcessor) RegRouter(method HTTP_METHOD, url string, handle HttpHandle){ func (slf *HttpSession) flush() {
pathRouter slf.w.WriteHeader(slf.statusCode)
if slf.msg!=nil {
slf.w.Write(slf.msg)
}
} }
func (slf *HttpProcessor) Router(session *HttpSession){ func (slf *HttpSession) done(){
slf.sessionDone <- slf
} }
func (slf *HttpService) SetHttpProcessor(httpProcessor IHttpProcessor) { func (slf *HttpSession) getMethod(method string) HTTP_METHOD {
slf.httpProcessor = httpProcessor switch method {
case "POST":
return METHOD_POST
case "GET":
return METHOD_GET
} }
func (slf *HttpService) Request(method HTTP_METHOD, url string, handle HttpHandle) error { return METHOD_INVALID
fnpath := runtime.FuncForPC(reflect.ValueOf(handle).Pointer()).Name()
sidx := strings.LastIndex(fnpath, "*")
if sidx == -1 {
return errors.New(fmt.Sprintf("http post func path is error, %s\n", fnpath))
} }
eidx := strings.LastIndex(fnpath, "-fm")
if sidx == -1 {
return errors.New(fmt.Sprintf("http post func path is error, %s\n", fnpath)) func (slf *HttpRouter) analysisRouterUrl(url string) (string, error) {
}
callpath := fnpath[sidx+1 : eidx] //替换所有空格
ridx := strings.LastIndex(callpath, ")") url = strings.ReplaceAll(url, " ", "")
if ridx == -1 { if len(url) <= 1 || url[0] != '/' {
return errors.New(fmt.Sprintf("http post func path is error, %s\n", fnpath)) return "", fmt.Errorf("url %s format is error!", url)
} }
hidx := strings.LastIndex(callpath, "HTTP_") //去掉尾部的/
if hidx == -1 { return strings.Trim(url, "/"), nil
return errors.New(fmt.Sprintf("http post func not contain HTTP_, %s\n", fnpath))
} }
callpath = strings.ReplaceAll(callpath, ")", "") func (slf *HttpSession) Handle(){
slf.httpRouter.Router(slf)
}
var r RouterMatchData func (slf *HttpRouter) SetFormFileKey(formFileKey string){
var matchURL string slf.formFileKey = formFileKey
var err error }
r.routerType = 0
r.callpath = "_" + callpath func (slf *HttpRouter) GetFormFileKey()string{
matchURL, err = AnalysisRouterUrl(url) return slf.formFileKey
}
func (slf *HttpRouter) PutHttpSession(httpSession *HttpSession){
slf.eventReciver.NotifyEvent(&event.Event{Type:event.Sys_Event_Http_Event,Data:httpSession})
}
func (slf *HttpRouter) RegRouter(method HTTP_METHOD, url string, handle HttpHandle) bool{
mapRouter,ok := slf.pathRouter[method]
if ok == false{
return false
}
mapRouter[strings.Trim(url,"/")] = routerMatchData{httpHandle:handle}
return true
}
func (slf *HttpRouter) Router(session *HttpSession){
if slf.httpFiltrateList!=nil {
for _,fun := range slf.httpFiltrateList{
if fun(session) == false {
session.done()
return
}
}
}
urlPath := session.GetPath()
for {
mapRouter, ok := slf.pathRouter[session.GetMethod()]
if ok == false {
break
}
v, ok := mapRouter[urlPath]
if ok == false {
break
}
v.httpHandle(session)
session.done()
return
}
for k, v := range slf.serveFileData {
idx := strings.Index(urlPath, k)
if idx != -1 {
session.fileData = v
session.done()
return
}
}
session.WriteStatusCode(http.StatusNotFound)
session.done()
}
func (slf *HttpService) SetHttpRouter(httpRouter IHttpRouter) {
slf.httpRouter = httpRouter
}
func (slf *HttpRouter) SetServeFile(method HTTP_METHOD, urlpath string, dirname string) error {
_, err := os.Stat(dirname)
if err != nil { if err != nil {
return err return err
} }
matchURL, aErr := slf.analysisRouterUrl(urlpath)
var strMethod string if aErr != nil {
if method == METHOD_GET { return aErr
strMethod = "GET"
} else if method == METHOD_POST {
strMethod = "POST"
} else {
return nil
} }
postAliasUrl[strMethod][matchURL] = r var routerData routerServeFileData
routerData.method = method
routerData.localPath = dirname
routerData.matchUrl = matchURL
slf.serveFileData[matchURL] = &routerData
return nil return nil
} }
func Post(url string, handle HttpHandle) error {
return Request(METHOD_POST, url, handle) type HttpFiltrate func(session *HttpSession) bool //true is pass
func (slf *HttpRouter) AddHttpFiltrate(FiltrateFun HttpFiltrate) bool {
slf.httpFiltrateList = append(slf.httpFiltrateList, FiltrateFun)
return false
} }
func Get(url string, handle HttpHandle) error { func (slf *HttpSession) Redirect(url string, cookieList []*http.Cookie) {
return Request(METHOD_GET, url, handle) redirectData := &HttpRedirectData{}
redirectData.Url = url
redirectData.CookieList = cookieList
slf.redirectData = redirectData
} }
func (slf *HttpSession) redirects() {
if slf.redirectData == nil {
return
}
if slf.redirectData.CookieList != nil {
for _, v := range slf.redirectData.CookieList {
http.SetCookie(slf.w, v)
}
}
http.Redirect(slf.w, slf.r, slf.redirectData.Url,
http.StatusTemporaryRedirect)
}
func (slf *HttpService) OnInit() error { func (slf *HttpService) OnInit() error {
slf.serverHTTPMux = ServeHTTPRouterMux{} iConfig := slf.GetServiceCfg()
slf.httpserver.Init(slf.port, &slf.serverHTTPMux, 10*time.Second, 10*time.Second) if iConfig == nil {
if slf.ishttps == true { return fmt.Errorf("%s service config is error!",slf.GetName())
slf.httpserver.SetHttps(slf.certfile, slf.keyfile)
} }
tcpCfg := iConfig.(map[string]interface{})
addr,ok := tcpCfg["ListenAddr"]
if ok == false {
return fmt.Errorf("%s service config is error!",slf.GetName())
}
var readTimeout time.Duration = Default_ReadTimeout
var writeTimeout time.Duration = Default_WriteTimeout
if cfgRead,ok := tcpCfg["ReadTimeout"];ok == true {
readTimeout = time.Duration(cfgRead.(float64))*time.Millisecond
}
if cfgWrite,ok := tcpCfg["WriteTimeout"];ok == true {
writeTimeout = time.Duration(cfgWrite.(float64))*time.Millisecond
}
slf.processTimeout = Default_ProcessTimeout
if cfgProcessTimeout,ok := tcpCfg["ProcessTimeout"];ok == true {
slf.processTimeout = time.Duration(cfgProcessTimeout.(float64))*time.Millisecond
}
slf.httpServer.Init(addr.(string), slf, readTimeout, writeTimeout)
//Set CAFile
caFileList,ok := tcpCfg["CAFile"]
if ok == false {
return nil
}
iCaList := caFileList.([]interface{})
var caFile [] network.CAFile
for _,i := range iCaList {
mapCAFile := i.(map[string]interface{})
c,ok := mapCAFile["Certfile"]
if ok == false{
continue
}
k,ok := mapCAFile["Certfile"]
if ok == false{
continue
}
if c.(string)!="" && k.(string)!="" {
caFile = append(caFile,network.CAFile{
Certfile: c.(string),
Keyfile: k.(string),
})
}
}
slf.httpServer.SetCAFile(caFile)
slf.httpServer.Start()
return nil return nil
} }
@@ -250,6 +385,60 @@ func (slf *HttpService) SetAlowOrigin(allowOrigin bool) {
slf.allowOrigin = allowOrigin slf.allowOrigin = allowOrigin
} }
func (slf *HttpService) ProcessFile(session *HttpSession){
upath := session.r.URL.Path
idx := strings.Index(upath, session.fileData.matchUrl)
subPath := strings.Trim(upath[idx+len(session.fileData.matchUrl):], "/")
destLocalPath := session.fileData.localPath + "/"+subPath
switch session.GetMethod() {
case METHOD_GET:
//判断文件夹是否存在
_, err := os.Stat(destLocalPath)
if err == nil {
http.ServeFile(session.w, session.r, destLocalPath)
} else {
session.WriteStatusCode(http.StatusNotFound)
session.flush()
return
}
//上传资源
case METHOD_POST:
// 在这儿处理例外路由接口
session.r.ParseMultipartForm(32 << 20) // max memory is set to 32MB
resourceFile, resourceFileHeader, err := session.r.FormFile(session.httpRouter.GetFormFileKey())
if err != nil {
session.WriteStatusCode(http.StatusNotFound)
session.flush()
return
}
defer resourceFile.Close()
//重新拼接文件名
imgFormat := strings.Split(resourceFileHeader.Filename, ".")
if len(imgFormat) < 2 {
session.WriteStatusCode(http.StatusNotFound)
session.flush()
return
}
filePrefixName := uuid.Rand().HexEx()
fileName := filePrefixName + "." + imgFormat[len(imgFormat)-1]
//创建文件
localpath := fmt.Sprintf("%s%s", destLocalPath, fileName)
localfd, err := os.OpenFile(localpath, os.O_WRONLY|os.O_CREATE, 0666)
if err != nil {
session.WriteStatusCode(http.StatusNotFound)
session.flush()
return
}
defer localfd.Close()
io.Copy(localfd, resourceFile)
session.WriteStatusCode(http.StatusOK)
session.Write([]byte(upath+"/"+fileName))
session.flush()
}
}
func (slf *HttpService) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (slf *HttpService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if slf.allowOrigin == true { if slf.allowOrigin == true {
if origin := r.Header.Get("Origin"); origin != "" { if origin := r.Header.Get("Origin"); origin != "" {
@@ -259,304 +448,37 @@ func (slf *HttpService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
"Action, Module") //有使用自定义头 需要这个,Action, Module是例子 "Action, Module") //有使用自定义头 需要这个,Action, Module是例子
} }
} }
if r.Method == "OPTIONS" { if r.Method == "OPTIONS" {
return return
} }
methodRouter, bok := postAliasUrl[r.Method] session := &HttpSession{sessionDone:make(chan *HttpSession,1),httpRouter:slf.httpRouter,statusCode:http.StatusOK}
if bok == false { session.r = r
writeRespone(w, http.StatusNotFound, fmt.Sprint("Can not support method.")) session.w = w
return
}
//权限验证
var errRet error
for _, filter := range slf.httpfiltrateList {
ret := filter(r.URL.Path, w, r)
if ret == nil {
errRet = nil
break
} else {
errRet = ret
}
}
if errRet != nil {
writeRespone(w, http.StatusOK, errRet.Error())
return
}
url := strings.Trim(r.URL.Path, "/")
var strCallPath string
matchData, ok := methodRouter[url]
if ok == true {
strCallPath = matchData.callpath
} else {
//如果是资源处理
for k, v := range staticRouterResource {
idx := strings.Index(url, k)
if idx != -1 {
staticServer(k, v, w, r)
return
}
}
// 拼接得到rpc服务的名称
vstr := strings.Split(url, "/")
if len(vstr) < 2 {
writeRespone(w, http.StatusNotFound, "Cannot find path.")
return
}
strCallPath = "_" + vstr[0] + ".HTTP_" + vstr[1]
}
defer r.Body.Close() defer r.Body.Close()
msg, err := ioutil.ReadAll(r.Body) body, err := ioutil.ReadAll(r.Body)
if err != nil { if err != nil {
writeRespone(w, http.StatusBadRequest, "") session.WriteStatusCode(http.StatusGatewayTimeout)
session.flush()
return return
} }
session.body = body
request := HttpRequest{r.Header, string(msg), r.URL.RawQuery, nil, r.URL.Path} slf.httpRouter.PutHttpSession(session)
var resp HttpRespone ticker := time.NewTicker(slf.processTimeout)
//resp.Resp = w select {
timeFuncStart := time.Now() case <-ticker.C:
err = cluster.InstanceClusterMgr().Call(strCallPath, &request, &resp) session.WriteStatusCode(http.StatusGatewayTimeout)
session.flush()
timeFuncPass := time.Since(timeFuncStart)
if bPrintRequestTime {
service.GetLogger().Printf(service.LEVER_INFO, "HttpServer Time : %s url : %s\n", timeFuncPass, strCallPath)
}
if err != nil {
writeRespone(w, http.StatusBadRequest, fmt.Sprint(err))
} else {
if resp.RedirectData.Url != "" {
resp.redirects(&w, r)
} else {
writeRespone(w, http.StatusOK, string(resp.Respone))
}
}
}
// CkResourceDir 检查静态资源文件夹路径
func SetStaticResource(method HTTP_METHOD, urlpath string, dirname string) error {
_, err := os.Stat(dirname)
if err != nil {
return err
}
matchURL, berr := AnalysisRouterUrl(urlpath)
if berr != nil {
return berr
}
var routerData RouterStaticResoutceData
if method == METHOD_GET {
routerData.method = "GET"
} else if method == METHOD_POST {
routerData.method = "POST"
} else {
return nil
}
routerData.localpath = dirname
staticRouterResource[matchURL] = routerData
return nil
}
func writeRespone(w http.ResponseWriter, status int, msg string) {
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
w.WriteHeader(status)
w.Write([]byte(msg))
}
type HttpFiltrate func(path string, w http.ResponseWriter, r *http.Request) error
func (slf *HttpService) AppendHttpFiltrate(fun HttpFiltrate) bool {
slf.serverHTTPMux.httpfiltrateList = append(slf.serverHTTPMux.httpfiltrateList, fun)
return false
}
func (slf *HttpService) OnRun() bool {
slf.httpserver.Start()
return false
}
func NewHttpServerService(port uint16) *HttpServerService {
http := new(HttpServerService)
http.port = port
return http
}
func (slf *HttpService) OnDestory() error {
return nil
}
func (slf *HttpService) OnSetupService(iservice service.IService) {
rpc.RegisterName(iservice.GetServiceName(), "HTTP_", iservice)
}
func (slf *HttpServerService) OnRemoveService(iservice service.IService) {
return
}
func (slf *HttpService) SetPrintRequestTime(isPrint bool) {
bPrintRequestTime = isPrint
}
func staticServer(routerUrl string, routerData RouterStaticResoutceData, w http.ResponseWriter, r *http.Request) {
upath := r.URL.Path
idx := strings.Index(upath, routerUrl)
subPath := strings.Trim(upath[idx+len(routerUrl):], "/")
destLocalPath := routerData.localpath + subPath
writeResp := func(status int, msg string) {
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
w.WriteHeader(status)
w.Write([]byte(msg))
}
switch r.Method {
//获取资源
case "GET":
//判断文件夹是否存在
_, err := os.Stat(destLocalPath)
if err == nil {
http.ServeFile(w, r, destLocalPath)
} else {
writeResp(http.StatusNotFound, "")
return
}
//上传资源
case "POST":
// 在这儿处理例外路由接口
/*
var errRet error
for _, filter := range slf.httpfiltrateList {
ret := filter(r.URL.Path, w, r)
if ret == nil {
errRet = nil
break break
case <- session.sessionDone:
if session.fileData!=nil {
slf.ProcessFile(session)
}else if session.redirectData!=nil {
session.redirects()
}else{ }else{
errRet = ret session.flush()
}
}
if errRet != nil {
w.Write([]byte(errRet.Error()))
return
}*/
r.ParseMultipartForm(32 << 20) // max memory is set to 32MB
resourceFile, resourceFileHeader, err := r.FormFile("file")
if err != nil {
fmt.Println(err)
writeResp(http.StatusNotFound, err.Error())
return
}
defer resourceFile.Close()
//重新拼接文件名
imgFormat := strings.Split(resourceFileHeader.Filename, ".")
if len(imgFormat) < 2 {
writeResp(http.StatusNotFound, "not a file")
return
}
filePrefixName := uuid.Rand().HexEx()
fileName := filePrefixName + "." + imgFormat[len(imgFormat)-1]
//创建文件
localpath := fmt.Sprintf("%s%s", destLocalPath, fileName)
localfd, err := os.OpenFile(localpath, os.O_WRONLY|os.O_CREATE, 0666)
if err != nil {
fmt.Println(err)
writeResp(http.StatusNotFound, "upload fail")
return
}
defer localfd.Close()
io.Copy(localfd, resourceFile)
writeResp(http.StatusOK, upath+"/"+fileName)
}
}
func (slf *HttpService) GetMethod(strCallPath string) (*reflect.Value, error) {
value, ok := slf.controllerMaps[strCallPath]
if ok == false {
err := fmt.Errorf("not find api")
return nil, err
}
return &value, nil
}
func (slf *HttpService) SetHttps(certfile string, keyfile string) bool {
if certfile == "" || keyfile == "" {
return false
}
slf.ishttps = true
slf.certfile = certfile
slf.keyfile = keyfile
return true
}
//序列化后写入Respone
func (slf *HttpRespone) WriteRespne(v interface{}) error {
StrRet, retErr := json.Marshal(v)
if retErr != nil {
slf.Respone = []byte(`{"Code": 2,"Message":"service error"}`)
service.GetLogger().Printf(sysmodule.LEVER_ERROR, "Json Marshal Error:%v\n", retErr)
} else {
slf.Respone = StrRet
}
return retErr
}
func (slf *HttpRespone) WriteRespones(Code int32, Msg string, Data interface{}) {
var StrRet string
//判断是否有错误码
if Code > 0 {
StrRet = fmt.Sprintf(`{"RCode": %d,"RMsg":"%s"}`, Code, Msg)
} else {
if Data == nil {
if Msg != "" {
StrRet = fmt.Sprintf(`{"RCode": 0,"RMsg":"%s"}`, Msg)
} else {
StrRet = `{"RCode": 0}`
}
} else {
if reflect.TypeOf(Data).Kind() == reflect.String {
StrRet = fmt.Sprintf(`{"RCode": %d , "Data": "%s"}`, Code, Data)
} else {
JsonRet, Err := json.Marshal(Data)
if Err != nil {
service.GetLogger().Printf(sysmodule.LEVER_ERROR, "common WriteRespone Json Marshal Err %+v", Data)
} else {
StrRet = fmt.Sprintf(`{"RCode": %d , "Data": %s}`, Code, JsonRet)
} }
} }
} }
}
slf.Respone = []byte(StrRet)
}
func (slf *HttpRespone) Redirect(url string, cookieList []*http.Cookie) {
slf.RedirectData.Url = url
slf.RedirectData.CookieList = cookieList
}
func (slf *HttpRespone) redirects(w *http.ResponseWriter, req *http.Request) {
if slf.RedirectData.CookieList != nil {
for _, v := range slf.RedirectData.CookieList {
http.SetCookie(*w, v)
}
}
http.Redirect(*w, req, slf.RedirectData.Url,
// see @andreiavrammsd comment: often 307 > 301
http.StatusTemporaryRedirect)
}