2015-04-26 19:33:35 +01:00
|
|
|
package broker
|
|
|
|
|
|
|
|
import (
|
|
|
|
"bytes"
|
|
|
|
"encoding/json"
|
|
|
|
"fmt"
|
|
|
|
"io/ioutil"
|
|
|
|
"net"
|
|
|
|
"net/http"
|
|
|
|
"strconv"
|
|
|
|
"strings"
|
|
|
|
"sync"
|
|
|
|
|
|
|
|
log "github.com/golang/glog"
|
2015-11-20 16:17:33 +00:00
|
|
|
"github.com/micro/go-micro/errors"
|
|
|
|
"github.com/micro/go-micro/registry"
|
2015-08-26 12:15:37 +01:00
|
|
|
"github.com/pborman/uuid"
|
2015-04-26 19:33:35 +01:00
|
|
|
)
|
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
type httpBroker struct {
|
2015-04-26 19:33:35 +01:00
|
|
|
id string
|
|
|
|
address string
|
2015-05-23 20:04:16 +01:00
|
|
|
unsubscribe chan *httpSubscriber
|
2015-04-26 19:33:35 +01:00
|
|
|
|
|
|
|
sync.RWMutex
|
2015-05-23 20:04:16 +01:00
|
|
|
subscribers map[string][]*httpSubscriber
|
2015-04-26 19:33:35 +01:00
|
|
|
running bool
|
|
|
|
exit chan chan error
|
|
|
|
}
|
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
type httpSubscriber struct {
|
2015-04-26 19:33:35 +01:00
|
|
|
id string
|
|
|
|
topic string
|
2015-05-23 20:04:16 +01:00
|
|
|
ch chan *httpSubscriber
|
2015-06-12 19:52:27 +01:00
|
|
|
fn Handler
|
2015-05-25 22:14:28 +01:00
|
|
|
svc *registry.Service
|
2015-04-26 19:33:35 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
var (
|
2015-05-23 20:04:16 +01:00
|
|
|
DefaultSubPath = "/_sub"
|
2015-04-26 19:33:35 +01:00
|
|
|
)
|
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
func newHttpBroker(addrs []string, opt ...Option) Broker {
|
|
|
|
addr := ":0"
|
2015-06-12 19:52:27 +01:00
|
|
|
if len(addrs) > 0 && len(addrs[0]) > 0 {
|
2015-05-23 20:04:16 +01:00
|
|
|
addr = addrs[0]
|
|
|
|
}
|
|
|
|
|
|
|
|
return &httpBroker{
|
2015-06-12 19:52:27 +01:00
|
|
|
id: "broker-" + uuid.NewUUID().String(),
|
2015-05-23 20:04:16 +01:00
|
|
|
address: addr,
|
|
|
|
subscribers: make(map[string][]*httpSubscriber),
|
|
|
|
unsubscribe: make(chan *httpSubscriber),
|
|
|
|
exit: make(chan chan error),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (h *httpSubscriber) Topic() string {
|
2015-04-26 19:33:35 +01:00
|
|
|
return h.topic
|
|
|
|
}
|
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
func (h *httpSubscriber) Unsubscribe() error {
|
2015-04-26 19:33:35 +01:00
|
|
|
h.ch <- h
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
func (h *httpBroker) start() error {
|
2015-04-26 19:33:35 +01:00
|
|
|
h.Lock()
|
|
|
|
defer h.Unlock()
|
|
|
|
|
|
|
|
if h.running {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
l, err := net.Listen("tcp", h.address)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
log.Infof("Broker Listening on %s", l.Addr().String())
|
|
|
|
h.address = l.Addr().String()
|
|
|
|
|
|
|
|
go http.Serve(l, h)
|
|
|
|
|
|
|
|
go func() {
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case ch := <-h.exit:
|
|
|
|
ch <- l.Close()
|
|
|
|
h.Lock()
|
|
|
|
h.running = false
|
|
|
|
h.Unlock()
|
|
|
|
return
|
|
|
|
case subscriber := <-h.unsubscribe:
|
|
|
|
h.Lock()
|
2015-05-23 20:04:16 +01:00
|
|
|
var subscribers []*httpSubscriber
|
2015-04-26 19:33:35 +01:00
|
|
|
for _, sub := range h.subscribers[subscriber.topic] {
|
|
|
|
if sub.id == subscriber.id {
|
|
|
|
registry.Deregister(sub.svc)
|
|
|
|
}
|
|
|
|
subscribers = append(subscribers, sub)
|
|
|
|
}
|
|
|
|
h.subscribers[subscriber.topic] = subscribers
|
|
|
|
h.Unlock()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
h.running = true
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
func (h *httpBroker) stop() error {
|
2015-04-26 19:33:35 +01:00
|
|
|
ch := make(chan error)
|
|
|
|
h.exit <- ch
|
|
|
|
return <-ch
|
|
|
|
}
|
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
func (h *httpBroker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
2015-04-26 19:33:35 +01:00
|
|
|
if req.Method != "POST" {
|
|
|
|
err := errors.BadRequest("go.micro.broker", "Method not allowed")
|
|
|
|
http.Error(w, err.Error(), http.StatusMethodNotAllowed)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
defer req.Body.Close()
|
|
|
|
|
|
|
|
b, err := ioutil.ReadAll(req.Body)
|
|
|
|
if err != nil {
|
|
|
|
errr := errors.InternalServerError("go.micro.broker", fmt.Sprintf("Error reading request body: %v", err))
|
|
|
|
w.WriteHeader(500)
|
|
|
|
w.Write([]byte(errr.Error()))
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2015-06-12 19:52:27 +01:00
|
|
|
var m *Message
|
|
|
|
if err = json.Unmarshal(b, &m); err != nil {
|
2015-04-26 19:33:35 +01:00
|
|
|
errr := errors.InternalServerError("go.micro.broker", fmt.Sprintf("Error parsing request body: %v", err))
|
|
|
|
w.WriteHeader(500)
|
|
|
|
w.Write([]byte(errr.Error()))
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2015-06-12 19:52:27 +01:00
|
|
|
topic := m.Header[":topic"]
|
|
|
|
delete(m.Header, ":topic")
|
|
|
|
|
|
|
|
if len(topic) == 0 {
|
2015-04-26 19:33:35 +01:00
|
|
|
errr := errors.InternalServerError("go.micro.broker", "Topic not found")
|
|
|
|
w.WriteHeader(500)
|
|
|
|
w.Write([]byte(errr.Error()))
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
h.RLock()
|
2015-06-12 19:52:27 +01:00
|
|
|
for _, subscriber := range h.subscribers[topic] {
|
|
|
|
subscriber.fn(m)
|
2015-04-26 19:33:35 +01:00
|
|
|
}
|
|
|
|
h.RUnlock()
|
|
|
|
}
|
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
func (h *httpBroker) Address() string {
|
2015-04-26 19:33:35 +01:00
|
|
|
return h.address
|
|
|
|
}
|
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
func (h *httpBroker) Connect() error {
|
2015-04-26 19:33:35 +01:00
|
|
|
return h.start()
|
|
|
|
}
|
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
func (h *httpBroker) Disconnect() error {
|
2015-04-26 19:33:35 +01:00
|
|
|
return h.stop()
|
|
|
|
}
|
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
func (h *httpBroker) Init() error {
|
2015-04-26 19:33:35 +01:00
|
|
|
if len(h.id) == 0 {
|
|
|
|
h.id = "broker-" + uuid.NewUUID().String()
|
|
|
|
}
|
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
http.Handle(DefaultSubPath, h)
|
2015-04-26 19:33:35 +01:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2015-06-12 19:52:27 +01:00
|
|
|
func (h *httpBroker) Publish(topic string, msg *Message) error {
|
2015-04-26 19:33:35 +01:00
|
|
|
s, err := registry.GetService("topic:" + topic)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2015-06-12 19:52:27 +01:00
|
|
|
msg.Header[":topic"] = topic
|
|
|
|
b, err := json.Marshal(msg)
|
2015-04-26 19:33:35 +01:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2015-11-08 01:48:48 +00:00
|
|
|
for _, service := range s {
|
|
|
|
for _, node := range service.Nodes {
|
|
|
|
r, err := http.Post(fmt.Sprintf("http://%s:%d%s", node.Address, node.Port, DefaultSubPath), "application/json", bytes.NewBuffer(b))
|
|
|
|
if err == nil {
|
|
|
|
r.Body.Close()
|
|
|
|
}
|
2015-04-26 19:33:35 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2015-06-12 19:52:27 +01:00
|
|
|
func (h *httpBroker) Subscribe(topic string, handler Handler) (Subscriber, error) {
|
2015-04-26 19:33:35 +01:00
|
|
|
// parse address for host, port
|
|
|
|
parts := strings.Split(h.Address(), ":")
|
|
|
|
host := strings.Join(parts[:len(parts)-1], ":")
|
|
|
|
port, _ := strconv.Atoi(parts[len(parts)-1])
|
|
|
|
|
|
|
|
// register service
|
2015-05-25 22:14:28 +01:00
|
|
|
node := ®istry.Node{
|
|
|
|
Id: h.id,
|
|
|
|
Address: host,
|
|
|
|
Port: port,
|
|
|
|
}
|
|
|
|
|
|
|
|
service := ®istry.Service{
|
|
|
|
Name: "topic:" + topic,
|
|
|
|
Nodes: []*registry.Node{node},
|
|
|
|
}
|
2015-04-26 19:33:35 +01:00
|
|
|
|
2015-05-23 20:04:16 +01:00
|
|
|
subscriber := &httpSubscriber{
|
2015-04-26 19:33:35 +01:00
|
|
|
id: uuid.NewUUID().String(),
|
|
|
|
topic: topic,
|
|
|
|
ch: h.unsubscribe,
|
2015-06-12 19:52:27 +01:00
|
|
|
fn: handler,
|
2015-04-26 19:33:35 +01:00
|
|
|
svc: service,
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := registry.Register(service); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
h.Lock()
|
|
|
|
h.subscribers[topic] = append(h.subscribers[topic], subscriber)
|
|
|
|
h.Unlock()
|
|
|
|
|
|
|
|
return subscriber, nil
|
|
|
|
}
|
2015-12-19 21:56:14 +00:00
|
|
|
|
|
|
|
func (h *httpBroker) String() string {
|
|
|
|
return "http"
|
|
|
|
}
|