Consider callback based server/client implementation

[thesyncim]

Callbacks are strictly more powerful and don’t require unnecessary goroutines. Also is easy to write correct code

Server implementation(untested)

type ServerV2 struct {
    Producer Producer

    Consumer Consumer
}
type MessageSender func(*Message) error

type Consumer interface {
    Init(topic, consumer string, sendMessage MessageSender) error
    ReceiveConfirmation(*Confirmation) error
    Close() error
}

type ConfirmationSender func(*Confirmation) error

type Producer interface {
    Init(topic string, publisher string, sendConfirmation ConfirmationSender) error
    ReceiveMessage(*Message) error
    Close() error
}

func (s *ServerV2) Consume(stream MessageSource_ConsumeServer) error {

    _, cancel := context.WithCancel(stream.Context())
    defer cancel()

    c := &consumer{Consumer: s.Consumer}
    for {
        msg, err := stream.Recv()
        if err != nil {
            if err == io.EOF {
                return nil
            }
            if strings.HasSuffix(err.Error(), "context canceled") {
                return nil
            }

            return err
        }
        //todo handle returned errors
        go c.handleRequest(msg, stream)
    }
}

type consumer struct {
    started  *atomicBool
    Consumer Consumer
}

func (c *consumer) handleRequest(msg *ConsumerRequest, stream MessageSource_ConsumeServer) error {
    switch {
    case msg.GetStartRequest() != nil:

        if c.started.Get() {
            return ErrStartedTwice
        }
        sr := msg.GetStartRequest()

        sendMessage:= func(m *Message) error {
            err := stream.Send(m)
            if err != nil {
                return err
            }
            return nil
        }
        if err := c.Consumer.Init(sr.GetTopic(), sr.GetConsumer(), sendMessage); err != nil {
            return err
        }
        c.started.Set(true)

    case msg.GetConfirmation() != nil:

        if !c.started.Get() {
            return ErrInvalidConfirm
        }

        return c.Consumer.ReceiveConfirmation(msg.GetConfirmation())
    default:
        return ErrInvalidRequest
    }
    panic("impossible")
}

type publisher struct {
    started  *atomicBool
    Producer Producer
}

func (p *publisher) handleRequest(msg *PublisherRequest, stream MessageSink_PublishServer) error {
    switch {
    case msg.GetStartRequest() != nil:
        if p.started.Get() {
            return ErrStartedTwice
        }

        sr := msg.GetStartRequest()
        sendConfirmation := func(c *Confirmation) error {
            return stream.Send(c)
        }
        p.Producer.Init(sr.Topic, "", sendConfirmation)
        p.started.Set(true)
    case msg.GetMsg() != nil:
        if !p.started.Get() {
            return ErrNotConnected
        }
        return p.Producer.ReceiveMessage(msg.GetMsg())

    }
    return nil
}

func (s *ServerV2) Publish(stream MessageSink_PublishServer) error {

    _, cancel := context.WithCancel(stream.Context())
    defer cancel()
    p := &publisher{Producer: s.Producer}
    for {
        msg, err := stream.Recv()
        if err != nil {
            if err == io.EOF {
                return err
            }
            if strings.HasSuffix(err.Error(), "context canceled") {
                return err
            }
            return err
        }
        //todo handle returned errors
        go p.handleRequest(msg, stream)
    }
}

type atomicBool struct{ flag int32 }

func (b *atomicBool) Set(value bool) {
    var i int32 = 0
    if value {
        i = 1
    }
    atomic.StoreInt32(&(b.flag), int32(i))
}

func (b *atomicBool) Get() bool {
    if atomic.LoadInt32(&(b.flag)) != 0 {
        return true
    }
    return false
}

and the backend(untested):

var _ = proximo.Consumer(&NatsHandlerConsumer{})
var _ = proximo.Producer(&NatsHandlerProducer{})

type NatsHandlerConsumer struct {
    Url string
    sub *nats.Subscription
}

func (n *NatsHandlerConsumer) Init(topic, consumer string, sendMessage proximo.MessageSender) error {
    conn, err := nats.Connect(n.Url)
    if err != nil {
        return err
    }
    //defer conn.Close()

    ch := make(chan *nats.Msg, 64) //TODO: make 64 configurable at startup time
    sub, err := conn.ChanSubscribe(topic, ch)
    if err != nil {
        return err
    }
    n.sub = sub

    for {
        select { // drop
        case m := <-ch:
            sendMessage(&proximo.Message{
                Data: m.Data,
                Id:   proximo.GenerateID(),
            })
        }
    }

    return nil
}

func (n *NatsHandlerConsumer) ReceiveConfirmation(c *proximo.Confirmation) error { return nil }
func (n *NatsHandlerConsumer) Close() error                                      { return n.sub.Unsubscribe() }

type NatsHandlerProducer struct {
    Url              string
    topic            string
    sendConfirmation proximo.ConfirmationSender
    conn             *nats.Conn
}

func (n *NatsHandlerProducer) Init(topic string, publisher string, sendConfirmation proximo.ConfirmationSender) error {
    var err error
    n.topic = topic
    n.conn, err = nats.Connect(n.Url)
    if err != nil {
        return err
    }

    return err
}

func (n *NatsHandlerProducer) ReceiveMessage(msg *proximo.Message) error {
    err := n.conn.Publish(n.topic, msg.GetData())
    n.sendConfirmation(&proximo.Confirmation{msg.GetId()})
    return err
}

func (n *NatsHandlerProducer) Close() error {
    n.conn.Close()
    return nil
}

[mjgarton]

Can you describe the problem that you are trying to solve please?

Or perhaps you have measured something as being slow due to excessive goroutines, in which case do you have benchmarks?

[thesyncim]

Sry if i was not clear. This is just a alternative way to solve the same problem. My main argument here (which is arguable) is an easy way to implement new backends as we can see in the example backend. Performance is not my motivation here. Just a collateral effect.