support verb headers

[flowchartsman]

I'm connecting to ActiveMQ using AckClient in an attempt to pull up to X messages and then ack the last one. Sadly, ActiveMQ requires a nonstandard activemq.prefetchSize header to be specified in the SUBSCRIBE headers for this behavior to work. Currently, the library doesn't seem to support this. Any way this could be a thing?

[ebenoist]

@alaska

I'm currently having a similar-ish issue, I've been using the following locally as a POC. Would this solve your issue as well?

// Subscribe creates a subscription on the STOMP server.
// The subscription has a destination, and messages sent to that destination
// will be received by this subscription. A subscription has a channel
// on which the calling program can receive messages.
func (c *Conn) Subscribe(destination string, ack AckMode) (*Subscription, error) {
    return c.SubscribeWithHeaders(destination, ack, NewHeader())
}

func (c *Conn) SubscribeWithHeaders(destination string, ack AckMode, userDefined *Header) (*Subscription, error) {
    ch := make(chan *Frame)
    headers := userDefined.Clone()

    if _, ok := headers.Contains(frame.Id); !ok {
        headers.Add(frame.Id, allocateId())
    }

    headers.Add(frame.Destination, destination)
    headers.Add(frame.Ack, ack.String())

    f := NewFrame(frame.SUBSCRIBE)
    f.Header = headers

    request := writeRequest{Frame: f, C: ch}

    sub := &Subscription{
        id:          headers.Get(frame.Id),
        destination: destination,
        conn:        c,
        ackMode:     ack,
        C:           make(chan *Message, 16),
    }
    go sub.readLoop(ch)

    c.writeCh <- request
    return sub, nil
}

[jjeffery]

I have been working on a new version of the API which solves this problem. Unfortunately the new API has some breaking changes from the old API, so this may be a problem for you.

The new API is documented at http://godoc.org/github.com/go-stomp/stomp -- there is now an ability to send custom header entries when you subscribe.

sub, err := conn.Subscribe("/queue/name", stomp.AckClient,
                        stomp.SuscribeOpt.Header("activemq.prefetchsize", "10))

if err != nil {
    return err
}

// and so on ..