Fix and comment broker service
This commit is contained in:
parent
c4b6d0f3a8
commit
04320d69ff
@ -324,15 +324,21 @@ func (h *httpBroker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
|||||||
p := &httpEvent{m: m, t: topic}
|
p := &httpEvent{m: m, t: topic}
|
||||||
id := req.Form.Get("id")
|
id := req.Form.Get("id")
|
||||||
|
|
||||||
|
var subs []Handler
|
||||||
|
|
||||||
h.RLock()
|
h.RLock()
|
||||||
for _, subscriber := range h.subscribers[topic] {
|
for _, subscriber := range h.subscribers[topic] {
|
||||||
if id == subscriber.id {
|
if id != subscriber.id {
|
||||||
// sub is sync; crufty rate limiting
|
continue
|
||||||
// so we don't hose the cpu
|
|
||||||
subscriber.fn(p)
|
|
||||||
}
|
}
|
||||||
|
subs = append(subs, subscriber.fn)
|
||||||
}
|
}
|
||||||
h.RUnlock()
|
h.RUnlock()
|
||||||
|
|
||||||
|
// execute the handler
|
||||||
|
for _, fn := range subs {
|
||||||
|
fn(p)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *httpBroker) Address() string {
|
func (h *httpBroker) Address() string {
|
||||||
@ -420,7 +426,6 @@ func (h *httpBroker) Connect() error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (h *httpBroker) Disconnect() error {
|
func (h *httpBroker) Disconnect() error {
|
||||||
|
|
||||||
h.RLock()
|
h.RLock()
|
||||||
if !h.running {
|
if !h.running {
|
||||||
h.RUnlock()
|
h.RUnlock()
|
||||||
|
@ -19,6 +19,7 @@ func (b *Broker) Publish(ctx context.Context, req *pb.PublishRequest, rsp *pb.Em
|
|||||||
Header: req.Message.Header,
|
Header: req.Message.Header,
|
||||||
Body: req.Message.Body,
|
Body: req.Message.Body,
|
||||||
})
|
})
|
||||||
|
log.Debugf("Published message to %s topic", req.Topic)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return errors.InternalServerError("go.micro.broker", err.Error())
|
return errors.InternalServerError("go.micro.broker", err.Error())
|
||||||
}
|
}
|
||||||
@ -49,12 +50,17 @@ func (b *Broker) Subscribe(ctx context.Context, req *pb.SubscribeRequest, stream
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return errors.InternalServerError("go.micro.broker", err.Error())
|
return errors.InternalServerError("go.micro.broker", err.Error())
|
||||||
}
|
}
|
||||||
defer sub.Unsubscribe()
|
defer func() {
|
||||||
|
log.Debugf("Unsubscribing from topic %s", req.Topic)
|
||||||
|
sub.Unsubscribe()
|
||||||
|
}()
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
|
log.Debugf("Context done for subscription to topic %s", req.Topic)
|
||||||
return nil
|
return nil
|
||||||
case <-errChan:
|
case err := <-errChan:
|
||||||
|
log.Debugf("Subscription error for topic %s: %v", req.Topic, err)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -3,7 +3,9 @@ package service
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/micro/go-micro/util/log"
|
||||||
"github.com/micro/go-micro/broker"
|
"github.com/micro/go-micro/broker"
|
||||||
pb "github.com/micro/go-micro/broker/service/proto"
|
pb "github.com/micro/go-micro/broker/service/proto"
|
||||||
"github.com/micro/go-micro/client"
|
"github.com/micro/go-micro/client"
|
||||||
@ -43,6 +45,7 @@ func (b *serviceBroker) Options() broker.Options {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (b *serviceBroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
|
func (b *serviceBroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
|
||||||
|
log.Debugf("Publishing to topic %s broker %v", topic, b.Addrs)
|
||||||
_, err := b.Client.Publish(context.TODO(), &pb.PublishRequest{
|
_, err := b.Client.Publish(context.TODO(), &pb.PublishRequest{
|
||||||
Topic: topic,
|
Topic: topic,
|
||||||
Message: &pb.Message{
|
Message: &pb.Message{
|
||||||
@ -58,6 +61,7 @@ func (b *serviceBroker) Subscribe(topic string, handler broker.Handler, opts ...
|
|||||||
for _, o := range opts {
|
for _, o := range opts {
|
||||||
o(&options)
|
o(&options)
|
||||||
}
|
}
|
||||||
|
log.Debugf("Subscribing to topic %s queue %s broker %v", topic, options.Queue, b.Addrs)
|
||||||
stream, err := b.Client.Subscribe(context.TODO(), &pb.SubscribeRequest{
|
stream, err := b.Client.Subscribe(context.TODO(), &pb.SubscribeRequest{
|
||||||
Topic: topic,
|
Topic: topic,
|
||||||
Queue: options.Queue,
|
Queue: options.Queue,
|
||||||
@ -74,7 +78,33 @@ func (b *serviceBroker) Subscribe(topic string, handler broker.Handler, opts ...
|
|||||||
closed: make(chan bool),
|
closed: make(chan bool),
|
||||||
options: options,
|
options: options,
|
||||||
}
|
}
|
||||||
go sub.run()
|
|
||||||
|
go func() {
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-sub.closed:
|
||||||
|
log.Debugf("Unsubscribed from topic %s", topic)
|
||||||
|
return
|
||||||
|
default:
|
||||||
|
// run the subscriber
|
||||||
|
log.Debugf("Streaming from broker %v to topic [%s] queue [%s]", b.Addrs, topic, options.Queue)
|
||||||
|
if err := sub.run(); err != nil {
|
||||||
|
log.Debugf("Resubscribing to topic %s broker %v", topic, b.Addrs)
|
||||||
|
stream, err := b.Client.Subscribe(context.TODO(), &pb.SubscribeRequest{
|
||||||
|
Topic: topic,
|
||||||
|
Queue: options.Queue,
|
||||||
|
}, client.WithAddress(b.Addrs...))
|
||||||
|
if err != nil {
|
||||||
|
log.Debugf("Failed to resubscribe to topic %s: %v", topic, err)
|
||||||
|
time.Sleep(time.Second)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// new stream
|
||||||
|
sub.stream = stream
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
return sub, nil
|
return sub, nil
|
||||||
}
|
}
|
||||||
|
@ -1,6 +1,7 @@
|
|||||||
package service
|
package service
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"github.com/micro/go-micro/util/log"
|
||||||
"github.com/micro/go-micro/broker"
|
"github.com/micro/go-micro/broker"
|
||||||
pb "github.com/micro/go-micro/broker/service/proto"
|
pb "github.com/micro/go-micro/broker/service/proto"
|
||||||
)
|
)
|
||||||
@ -31,24 +32,45 @@ func (s *serviceEvent) Ack() error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *serviceSub) run() {
|
func (s *serviceSub) isClosed() bool {
|
||||||
|
select {
|
||||||
|
case <-s.closed:
|
||||||
|
return true
|
||||||
|
default:
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *serviceSub) run() error {
|
||||||
exit := make(chan bool)
|
exit := make(chan bool)
|
||||||
go func() {
|
go func() {
|
||||||
select {
|
select {
|
||||||
case <-exit:
|
case <-exit:
|
||||||
return
|
|
||||||
case <-s.closed:
|
case <-s.closed:
|
||||||
s.stream.Close()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// close the stream
|
||||||
|
s.stream.Close()
|
||||||
}()
|
}()
|
||||||
|
|
||||||
for {
|
for {
|
||||||
// TODO: do not fail silently
|
// TODO: do not fail silently
|
||||||
msg, err := s.stream.Recv()
|
msg, err := s.stream.Recv()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
log.Debugf("Streaming error for subcription to topic %s: %v", s.Topic(), err)
|
||||||
|
|
||||||
|
// close the exit channel
|
||||||
close(exit)
|
close(exit)
|
||||||
return
|
|
||||||
|
// don't return an error if we unsubscribed
|
||||||
|
if s.isClosed() {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// return stream error
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: handle error
|
// TODO: handle error
|
||||||
s.handler(&serviceEvent{
|
s.handler(&serviceEvent{
|
||||||
topic: s.topic,
|
topic: s.topic,
|
||||||
|
Loading…
Reference in New Issue
Block a user