micro/broker/service/handler/handler.go

67 lines
1.6 KiB
Go
Raw Normal View History

2019-10-03 18:19:02 +03:00
package handler
import (
"context"
"github.com/micro/go-micro/broker"
pb "github.com/micro/go-micro/broker/service/proto"
"github.com/micro/go-micro/errors"
2019-10-04 18:40:16 +03:00
"github.com/micro/go-micro/util/log"
2019-10-03 18:19:02 +03:00
)
type Broker struct {
Broker broker.Broker
}
func (b *Broker) Publish(ctx context.Context, req *pb.PublishRequest, rsp *pb.Empty) error {
2019-10-03 19:30:37 +03:00
log.Debugf("Publishing message to %s topic", req.Topic)
2019-10-03 18:19:02 +03:00
err := b.Broker.Publish(req.Topic, &broker.Message{
Header: req.Message.Header,
Body: req.Message.Body,
})
2019-10-04 18:30:03 +03:00
log.Debugf("Published message to %s topic", req.Topic)
2019-10-03 18:19:02 +03:00
if err != nil {
return errors.InternalServerError("go.micro.broker", err.Error())
}
return nil
}
func (b *Broker) Subscribe(ctx context.Context, req *pb.SubscribeRequest, stream pb.Broker_SubscribeStream) error {
errChan := make(chan error, 1)
// message handler to stream back messages from broker
handler := func(p broker.Event) error {
if err := stream.Send(&pb.Message{
Header: p.Message().Header,
Body: p.Message().Body,
}); err != nil {
select {
case errChan <- err:
return err
default:
return err
}
}
return nil
}
2019-10-03 19:30:37 +03:00
log.Debugf("Subscribing to %s topic", req.Topic)
2019-10-03 18:19:02 +03:00
sub, err := b.Broker.Subscribe(req.Topic, handler, broker.Queue(req.Queue))
if err != nil {
return errors.InternalServerError("go.micro.broker", err.Error())
}
2019-10-04 18:30:03 +03:00
defer func() {
log.Debugf("Unsubscribing from topic %s", req.Topic)
sub.Unsubscribe()
}()
2019-10-03 18:19:02 +03:00
select {
case <-ctx.Done():
2019-10-04 18:30:03 +03:00
log.Debugf("Context done for subscription to topic %s", req.Topic)
2019-10-03 18:19:02 +03:00
return nil
2019-10-04 18:30:03 +03:00
case err := <-errChan:
log.Debugf("Subscription error for topic %s: %v", req.Topic, err)
2019-10-03 18:19:02 +03:00
return err
}
}