consul init

common/url.go

@@ -201,16 +201,13 @@ func NewURL(ctx context.Context, urlString string, opts ...option) (URL, error)
 	for _, opt := range opts {
 		opt(&s)
 	}
-	//fmt.Println(s.String())
+
 	return s, nil
 }
 
-//
-//func (c URL) Key() string {
-//	return fmt.Sprintf(
-//		"%s://%s:%s@%s:%s/%s",
-//		c.Protocol, c.Username, c.Password, c.Ip, c.Port, c.Path)
-//}
+func NewURLFromString(url string) URL {
+
+}
 
 func (c URL) URLEqual(url URL) bool {
 	c.Ip = ""

registry/consul/listener.go

+package consul
+
+import (
+	consul "github.com/hashicorp/consul/api"
+	"github.com/hashicorp/consul/watch"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/registry"
+)
+
+type consulListener struct {
+	plan      *watch.Plan
+	addrsChan chan []*consul.ServiceEntry
+}
+
+func newConsulListener(url common.URL) (registry.Listener, error) {
+	var err error
+
+	addrsChan := make(chan []*consul.ServiceEntry, 1)
+
+	params := make(map[string]interface{})
+	params["type"] = "service"
+	params["service"] = url.Service()
+	plan, err := watch.Parse(params)
+	if err != nil {
+		return nil, err
+	}
+	plan.Handler = func(idx uint64, raw interface{}) {
+		addrs, _ := raw.([]*consul.ServiceEntry)
+		addrsChan <- addrs
+	}
+
+	listener := &consulListener{
+		plan:      plan,
+		addrsChan: addrsChan,
+	}
+	return listener, nil
+}
+
+func (l *consulListener) Next() (*registry.ServiceEvent, error) {
+	return nil, nil
+}
+
+func (l *consulListener) Close() {
+
+}
\ No newline at end of file

registry/consul/registry.go

+package consul
+
+import (
+	"strconv"
+)
+
+import (
+	perrors "github.com/pkg/errors"
+	consul "github.com/hashicorp/consul/api"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/registry"
+	"github.com/apache/dubbo-go/common/constant"
+)
+
+type consulRegistry struct {
+	*common.URL
+	client *consul.Client
+}
+
+func newConsulRegistry(url *common.URL) (registry.Registry, error) {
+	var err error
+
+	config := &consul.Config{Address: url.Location}
+	client, err := consul.NewClient(config)
+	if err != nil {
+		return nil, err
+	}
+
+	r := &consulRegistry{
+		URL:    url,
+		client: client,
+	}
+
+	return r, nil
+}
+
+func (r *consulRegistry) Register(url common.URL) error {
+	var err error
+
+	role, _ := strconv.Atoi(r.URL.GetParam(constant.ROLE_KEY, ""))
+	if role == common.PROVIDER {
+		err = r.register(url)
+		if err != nil {
+			return perrors.WithStack(err)
+		}
+	}
+	return nil
+}
+
+func (r *consulRegistry) register(url common.URL) error {
+	var err error
+
+	service, err := buildService(url)
+	if err != nil {
+		return err
+	}
+	return r.client.Agent().ServiceRegister(service)
+}
+
+func (r *consulRegistry) Subscribe(url common.URL) (registry.Listener, error) {
+	var (
+		listener registry.Listener
+		err 	 error
+	)
+
+	role, _ := strconv.Atoi(r.URL.GetParam(constant.ROLE_KEY, ""))
+	if role == common.CONSUMER {
+		listener, err = r.subscribe(url)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return listener, nil
+}
+
+func (r *consulRegistry) subscribe(url common.URL) (registry.Listener, error) {
+	var err error
+
+	listener, err := newConsulListener(url)
+	return listener, err
+}
+
+func (r *consulRegistry) GetUrl() common.URL {
+	return *r.URL
+}
+
+func (r *consulRegistry) IsAvailable() bool {
+
+}
+
+func (r *consulRegistry) Destroy() {
+
+}

registry/consul/utils.go

+package consul
+
+import (
+	"strconv"
+	"crypto/md5"
+)
+
+import (
+	consul "github.com/hashicorp/consul/api"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+)
+
+func buildId(url common.URL) string {
+	return string(md5.Sum([]byte(url.String()))[:])
+}
+
+func buildService(url common.URL) (*consul.AgentServiceRegistration, error) {
+	var err error
+
+	// id
+	id := buildId(url)
+
+	// port
+	port, err := strconv.Atoi(url.Port)
+	if err != nil {
+		return nil, err
+	}
+
+	// tags
+	tags := make([]string, 0)
+	for k := range url.Params {
+		tags = append(tags, k + "=" + url.Params.Get(k))
+	}
+
+	// meta
+	meta := make(map[string]string)
+	meta["url"] = url.String()
+
+	service := &consul.AgentServiceRegistration{
+		Name:    url.Service(),
+		ID:      id,
+		Address: url.Ip,
+		Port:    port,
+		Tags:    tags,
+		Meta:    meta,
+	}
+
+	return service, nil
+}
+
+func retrieveURL(service *consul.ServiceEntry) common.URL {
+	url := service.Service.Meta["url"]
+	return common.NewURLFromString(url)
+}