Merge pull request #97 from weibocom/dev

Dev

Author: rayzhang0603

agent.go

@@ -1,7 +1,6 @@
 package motan
 
 import (
-	"encoding/json"
 	"flag"
 	"fmt"
 	"io/ioutil"
@@ -27,7 +26,7 @@ const (
 
 type Agent struct {
 	ConfigFile string
-	extFactory motan.ExtentionFactory
+	extFactory motan.ExtensionFactory
 	Context    *motan.Context
 
 	agentServer motan.Server
@@ -47,10 +46,10 @@ type Agent struct {
 	manageHandlers map[string]http.Handler
 }
 
-func NewAgent(extfactory motan.ExtentionFactory) *Agent {
+func NewAgent(extfactory motan.ExtensionFactory) *Agent {
 	var agent *Agent
 	if extfactory == nil {
-		fmt.Println("agent using default extentionFactory.")
+		fmt.Println("agent using default extensionFactory.")
 		agent = &Agent{extFactory: GetDefaultExtFactory()}
 	} else {
 		agent = &Agent{extFactory: extfactory}
@@ -106,6 +105,7 @@ func (a *Agent) initParam() {
 		logdir = "."
 	}
 	initLog(logdir)
+	registerSwitchers(a.Context)
 
 	port := *motan.Port
 	if port == 0 && section != nil && section["port"] != nil {
@@ -165,7 +165,7 @@ func (a *Agent) SetSanpshotConf() {
 	if snapshotDir == "" {
 		snapshotDir = registry.DefaultSnapshotDir
 	}
-	registry.SetSanpshotConf(registry.DefaultSnapshotInterval, snapshotDir)
+	registry.SetSnapshotConf(registry.DefaultSnapshotInterval, snapshotDir)
 }
 
 func (a *Agent) initAgentURL() {
@@ -215,7 +215,7 @@ func (a *Agent) startAgent() {
 }
 
 func (a *Agent) registerAgent() {
-	vlog.Infoln("start agent regitstry.")
+	vlog.Infoln("start agent registry.")
 	if reg, exit := a.agentURL.Parameters[motan.RegistryKey]; exit {
 		if registryURL, regexit := a.Context.RegistryURLs[reg]; regexit {
 			registry := a.extFactory.GetRegistry(registryURL)
@@ -232,7 +232,7 @@ func (a *Agent) registerAgent() {
 				}
 			}
 		} else {
-			vlog.Warningf("can not find agent registry in conf, so do not register. agent url:%s\n", a.agentURL)
+			vlog.Warningf("can not find agent registry in conf, so do not register. agent url:%s\n", a.agentURL.GetIdentity())
 		}
 	}
 }
@@ -288,6 +288,7 @@ func (a *Agent) startServerAgent() {
 			application = a.agentURL.GetParam(motan.ApplicationKey, "")
 			url.PutParam(motan.ApplicationKey, application)
 		}
+		url.ClearCachedInfo()
 		exporter := &mserver.DefaultExporter{}
 		provider := a.extFactory.GetProvider(url)
 		if provider == nil {
@@ -296,7 +297,7 @@ func (a *Agent) startServerAgent() {
 		}
 		motan.CanSetContext(provider, globalContext)
 		motan.Initialize(provider)
-		provider = mserver.WarperWithFilter(provider, a.extFactory)
+		provider = mserver.WrapWithFilter(provider, a.extFactory, globalContext)
 		exporter.SetProvider(provider)
 		server := a.agentPortServer[url.Port]
 		if server == nil {
@@ -373,6 +374,14 @@ func initLog(logdir string) {
 	vlog.LogInit(nil)
 }
 
+func registerSwitchers(c *motan.Context) {
+	switchers, _ := c.Config.GetSection(motan.SwitcherSection)
+	s := motan.GetSwitcherManager()
+	for n, v := range switchers {
+		s.Register(n.(string), v.(bool))
+	}
+}
+
 func getDefaultResponse(requestid uint64, errmsg string) *motan.MotanResponse {
 	return motan.BuildExceptionResponse(requestid, &motan.Exception{ErrCode: 400, ErrMsg: errmsg, ErrType: motan.ServiceException})
 }
@@ -410,24 +419,15 @@ func (a *Agent) RegisterManageHandler(path string, handler http.Handler) {
 }
 
 func (a *Agent) startMServer() {
-	if _, ok := a.manageHandlers["/"]; !ok {
-		a.manageHandlers["/"] = http.HandlerFunc(a.rootHandler)
-	}
-	if _, ok := a.manageHandlers["/503"]; !ok {
-		a.manageHandlers["/503"] = http.HandlerFunc(a.StatusChangeHandler)
-	}
-	if _, ok := a.manageHandlers["/200"]; !ok {
-		a.manageHandlers["/200"] = http.HandlerFunc(a.StatusChangeHandler)
-	}
-	if _, ok := a.manageHandlers["/getConfig"]; !ok {
-		a.manageHandlers["/getConfig"] = http.HandlerFunc(a.getConfigHandler)
-	}
-	if _, ok := a.manageHandlers["/getReferService"]; !ok {
-		a.manageHandlers["/getReferService"] = http.HandlerFunc(a.getReferServiceHandler)
+	handlers := make(map[string]http.Handler, 16)
+	for k, v := range GetDefaultManageHandlers() {
+		handlers[k] = v
 	}
 	for k, v := range a.manageHandlers {
-		http.Handle(k, v)
-		vlog.Infof("add manage server handle path:%s\n", k)
+		handlers[k] = v
+	}
+	for k, v := range handlers {
+		a.mhandle(k, v)
 	}
 
 	vlog.Infof("start listen manage port %d ...\n", a.mport)
@@ -438,60 +438,31 @@ func (a *Agent) startMServer() {
 	}
 }
 
-type rpcService struct {
-	Name   string `json:"name"`
-	Status bool   `json:"status"`
-}
-
-type body struct {
-	Service []rpcService `json:"service"`
-}
-
-type jsonRetData struct {
-	Code int  `json:"code"`
-	Body body `json:"body"`
-}
-
-// return agent server status, e.g. 200 or 503
-func (a *Agent) rootHandler(w http.ResponseWriter, r *http.Request) {
-	w.WriteHeader(a.status)
-	w.Write([]byte(http.StatusText(a.status)))
-}
-
-func (a *Agent) getConfigHandler(w http.ResponseWriter, r *http.Request) {
-	data, err := ioutil.ReadFile(*motan.CfgFile)
-	if err != nil {
-		w.Write([]byte("error."))
-	} else {
-		w.Write(data)
-	}
-}
-
-func (a *Agent) getReferServiceHandler(w http.ResponseWriter, r *http.Request) {
-
-	mbody := body{Service: []rpcService{}}
-	for _, cls := range a.clustermap {
-		rpc := cls.GetURL().Path
-		available := cls.IsAvailable()
-		mbody.Service = append(mbody.Service, rpcService{Name: rpc, Status: available})
-	}
-	retData := &jsonRetData{Code: 200, Body: mbody}
-	if data, err := json.Marshal(&retData); err == nil {
-		w.Write(data)
-	} else {
-		w.Write([]byte("error."))
+func (a *Agent) mhandle(k string, h http.Handler) {
+	defer func() {
+		if err := recover(); err != nil {
+			vlog.Warningf("manageHandler register fail. maybe the pattern '%s' already registered\n", k)
+		}
+	}()
+	if sa, ok := h.(SetAgent); ok {
+		sa.SetAgent(a)
 	}
+	http.HandleFunc(k, func(w http.ResponseWriter, r *http.Request) {
+		if !PermissionCheck(r) {
+			w.Write([]byte("need permission!"))
+			return
+		}
+		defer func() {
+			if err := recover(); err != nil {
+				fmt.Fprintf(w, "process request err: %s\n", err)
+			}
+		}()
+		h.ServeHTTP(w, r)
+	})
+	vlog.Infof("add manage server handle path:%s\n", k)
 }
 
-// StatusChangeHandler change agent server status, and set registed services available or unavailable.
-func (a *Agent) StatusChangeHandler(w http.ResponseWriter, r *http.Request) {
-	switch r.RequestURI {
-	case "/200":
-		availableService(a.serviceRegistries)
-		a.status = http.StatusOK
-	case "/503":
-		unavailableService(a.serviceRegistries)
-		a.status = http.StatusServiceUnavailable
-	}
-	w.Write([]byte("ok."))
+func (a *Agent) getConfigData() []byte {
+	data, _ := ioutil.ReadFile(*motan.CfgFile)
+	return data
 }

client.go

@@ -19,7 +19,7 @@ var (
 type MCContext struct {
 	confFile   string
 	context    *motan.Context
-	extFactory motan.ExtentionFactory
+	extFactory motan.ExtensionFactory
 	clients    map[string]*Client
 
 	csync  sync.Mutex
@@ -29,7 +29,7 @@ type MCContext struct {
 type Client struct {
 	url        *motan.URL
 	cluster    *cluster.MotanCluster
-	extFactory motan.ExtentionFactory
+	extFactory motan.ExtensionFactory
 }
 
 func (c *Client) Call(method string, args []interface{}, reply interface{}) error {
@@ -111,6 +111,7 @@ func GetClientContext(confFile string) *MCContext {
 			logdir = "."
 		}
 		initLog(logdir)
+		registerSwitchers(mc.context)
 	}
 	return mc
 }
@@ -127,7 +128,7 @@ func (m *MCContext) Initialize() {
 	}
 }
 
-func (m *MCContext) Start(extfactory motan.ExtentionFactory) {
+func (m *MCContext) Start(extfactory motan.ExtensionFactory) {
 	m.csync.Lock()
 	defer m.csync.Unlock()
 	m.extFactory = extfactory