gmallard / stompngo Goto Github PK

Before the call to:

transmitCommon // this does headers.Clone()

in modules:

abort.go
ack.go
begin.go
commit.go
nack.go

Per our discussion in stompngo_examples issue #2, I'm opening an issue in the main project. The concern is on what to do with in-flight messages received after or during an unsubscribe. The current use case I've been working with involves ActiveMQ with prefetch set, but could apply to any similar interaction with a STOMP server that sends more than one MESSAGE without requiring an ACK using either ackmode client or client-individual

1. connect
2. sub to queue q.1 with ackmode client and prefetch X where X is how many messages I need
3. process X messages
4. ACK the last message (which should be cumulative with ackmode client)
5. unsubscribe
6. repeat 2-5 with queues q.2, q.3, etc.

The problem I'm running into is that the UNSUBSCRIBE right after the ACK seems to be causing messages to not get ACKed sometimes, and some of the later subscriptions are getting nothing at all. In examining this problem (which may or may not be related to the issue I'm submitting), I ran across the following scenario which I think needs to be at least mentioned in the documentation, if not fixed in the code somehow:

1. Client subscribes to a destination, which creates a chan MessageData with a buffer size of c.scc (default: 1)

//subscription.go
if hid { // Client specified id
     c.subs[id] = make(chan MessageData, c.scc) // Assign subscription
}

2. Message(s) arrive with that destination specified in the subscription header and are picked up in the read loop here

//reader.go
if sid, ok := f.Headers.Contains("subscription"); ok {
     c.subsLock.Lock()
     c.subs[sid] <- d
     c.subsLock.Unlock()
} else {
     c.input <- d
}

3. Client processes some messages or not, but there is still a message in the channel for this subscription.
4. client unsubscribes, which causes the following

//unsubscribe.go
c.subsLock.Lock()                                                                                                                                                              defer c.subsLock.Unlock()
// ...
close(c.subs[sid])
delete(c.subs, sid)

5. Before c.subsLock.Lock() is called in the unsubscribe, another message arrives and is picked up by the reader goroutine.

It seems that this would block. The unsubscribe wouldn't be able to get the subsLock, and the reader goroutine would be blocked trying to send to the subscription channel.

We had a service that was being disrupted by a panic on the heartbeating logic:

Heartbeat Send Failure: write tcp 172.22.0.88:61613: connection reset by peer
panic: non-positive interval for NewTicker

goroutine 351 [running]:
runtime.panic(0x7746c0, 0xc2091ce2f0)
    /usr/local/go/src/pkg/runtime/panic.c:279 +0xf5
time.NewTicker(0xffffffffcde10e9d, 0x62082dc)
    /usr/local/go/src/pkg/time/tick.go:23 +0xd8
github.com/gmallard/stompngo.(*Connection).sendTicker(0xc2080ba500)
    /srv/teamcity/work/7d0e7d4c05ce72c/src/github.com/monsooncommerce/last-offer-event-storage-service/Godeps/_workspace/src/github.com/gmallard/stompngo/heartbeats.go:122 +0x7b
created by github.com/gmallard/stompngo.(*Connection).initializeHeartBeats
    /srv/teamcity/work/7d0e7d4c05ce72c/src/github.com/monsooncommerce/last-offer-event-storage-service/Godeps/_workspace/src/github.com/gmallard/stompngo/heartbeats.go:100 +0x9b4

the timing bounds set in the previous heartbeat were creating a negative duration which caused a panic.

I have submitted a pull request that solves this by limiting the outstanding heartbeats to 10, and refactoring the time interval logic:
#22

The package as currently implemented does not support CRLF line ends as required for 1.2 level connections.

The check for this spec requirement is not currently implemented. Servers might catch this, but it needs to be done before the CONNECT goes to the server.

In util.go::readBody if the underlying (TCP) connection over which STOMP is being communicated unexpectedly closes, with an error like

read tcp 10.3.65.36:44519->52.43.31.97:1443: use of closed network connection

the reader will yield zero bytes, causing the following statement to panic since n == 0:

if n < l { // Short read, e is ErrUnexpectedEOF
    c.log("SHORT READ", n, l, e)
    return b[0 : n-1], e  // panic: runtime error: slice bounds out of range
}

Hi @gmallard

stompngo is a awesome project. I wonder is there any other use case in production environment. We have put it under production environment with rabbitmq and rabbitmq-stomp-websocket plugin, which can provide us a channel to push message from server to agents under a web app like connection.

Hi, by default github and godoc are displaying the master branch as default, but when calling go get github.com/gmallard/stompngo a go1 branch is downloaded.

Which branch should I use? Master seems more reasonable, the missing constants in the go1 branch like HK_DESTINATION are annoying.

The package should implement:

• Heartbeat send counter
• Heartbeat receive counter

And should expose to the client:

• Heartbeat send counter
• Heartbeat receive counter
• Heartbeat send interval (in ms)
• Heartbeat receive interval (in ms)

After a successful connection:

fmt.Println("bodylen", len(conn.ConnectResponse.Body))

shows 1. This is erroneous, the value should be 0.

The package currently does not support the following use case.

• A STOMP 1.2 client sends messages to a broker.
• These messages have headers that require the encode / decode process.
• A STOMP 1.0 client connects to the broker, and subscribes
• The messages received will have encoded headers
• We are assuming that the 1.0 client wants decoded headers.

Solution: always decode headers when received.

Hi,
Thanks a lot for this really nice library :)
I've got a small question.
Do I understand correctly that the library automatically takes care of heart-beating?
For example, in the documentation of Disconnect, it says

Disconnect from a STOMP broker.

Shut down heart beats if necessary. Set connection status to false to disable further actions with this connection.

Is the "Shut down" meant as an imperative or a description of what the function does? ("Disconnect" in the first sentence seems to be a description, so I'm not sure)

Thanks a lot for your help in advance!

The current library does not handle empty header names and values properly.

Empty names and values should be handled per the Stomp specs:

• For 1.0 empty names are not allowed, and empty headers are not allowed
• For 1.1 and 1.2 empty names are not allowed (but empty values are)

Desired behavior on error (input headers from client or broker):

• Return a error to the client
• Shutdown the connection

Not sure how reconnection or detection of failed network connections is supposed to work.. .but...

reader.go contains a bug in which it does not properly detect EOF.

This may be by design, however if you close(c) after the for {} loop, the user and use the coma, ok ("msg, ok := <-r") pattern to detect when the channel is closed and attempt a reconnect.

The fix is to close the input channel which will signal a closed channel to the receiver.

~ line 70 of func (c *Connection) reader() {

70 fmt.Println("Dead connection")
71 close(c.input)

Reconnection could be made automatic but it might be up to the caller to handle this also.

When Disconnect is called and a receipt is requested, the code reads a MessageData structure from the conn.input channel.

This MessageData structure could:

a) Contain an ERROR frame
b) Contain a RECEIPT frame, but for the wrong receipt

Disconnect should check for both of these conditions, and report them to the caller with a non-nil error.

Disconnect can block forever on line https://github.com/gmallard/stompngo/blob/dev/disconnect.go#L69 if the reader is shutdown before the disconnect receipt has been read.

This can be shown by adding a sleep in reader (TestDiscRace will hang):

diff --git a/conndisc_test.go b/conndisc_test.go
index 1e5527d..bc95097 100644
--- a/conndisc_test.go
+++ b/conndisc_test.go
@@ -306,3 +306,22 @@ func TestConnRespData(t *testing.T) {
                }
        }
 }
+
+func TestDiscRace(t *testing.T) {
+       t.Parallel()
+       n, _ := openConn(t)
+       ch := check11(TEST_HEADERS)
+       c, _ := Connect(n, ch)
+
+       r := "my-receipt-001"
+       e := c.Send(Headers{"destination", "my-dest", "receipt", r},
+               "my-data")
+       if e != nil {
+               t.Errorf("Send: got error %v\n", e)
+       }
+       _ = <-c.MessageData
+
+       _ = c.Disconnect(empty_headers)
+       _ = closeConn(t, n)
+
+}
diff --git a/reader.go b/reader.go
index 46ecc58..3ce0f60 100644
--- a/reader.go
+++ b/reader.go
@@ -60,6 +60,8 @@ func (c *Connection) reader() {

                c.log("RECEIVE", m.Command, m.Headers)

+               time.Sleep(time.Second)
+
                select {
                case q = <-c.rsd:
                default:

Suggested fix: shut down the reader later or close the input channel when the reader exits. or both.

While connecting to an Apache Apollo broker I ran into the following:

panic: runtime error: index out of range

goroutine 7 [running]:
runtime.panic(0x6292c0, 0x8e4837)
    /usr/local/go/src/pkg/runtime/panic.c:266 +0xb6
github.com/gmallard/stompngo.(*Connection).readFrame(0xc210072000, 0xc21000aa60, 0xe, 0x8f2490, 0x0, ...)
    /Users/kelseyhightower/go/src/github.com/gmallard/stompngo/reader.go:122 +0x85e
github.com/gmallard/stompngo.(*Connection).reader(0xc210072000)
    /Users/kelseyhightower/go/src/github.com/gmallard/stompngo/reader.go:36 +0x3a
created by github.com/gmallard/stompngo.Connect
    /Users/kelseyhightower/go/src/github.com/gmallard/stompngo/connect.go:111 +0x7c9

The root cause seems to be reading an invalid header during the readFrame() method:

s = s[0 : len(s)-1]
p := strings.SplitN(s, ":", 2)
if c.Protocol() != SPL_10 && f.Command != CONNECTED {
  p[0] = decode(p[0])
  p[1] = decode(p[1])
}

which leads to

panic: runtime error: index out of range

I'm thinking we need to do some bounds checking here, and return a custom error message regrading the invalid headers. I'm happy to work on this if you think this is the right direction to go.

dmiles@ixtab:/go/src/github.com/danieltmiles/stompngo$ go version
go version go1.3.1 linux/amd64
dmiles@ixtab:/go/src/github.com/danieltmiles/stompngo$ env | grep GO
GOPATH=/home/dmiles/go
dmiles@ixtab:/go/src/github.com/danieltmiles/stompngo$ go build
dmiles@ixtab:/go/src/github.com/danieltmiles/stompngo$ echo $?
0
dmiles@ixtab:~/go/src/github.com/danieltmiles/stompngo$ go test -v
=== RUN TestAckErrors
=== RUN TestAckSameConn
=== RUN TestAckDiffConn
=== RUN TestCodecEncodeBasic
=== RUN TestCodecDecodeBasic
=== RUN TestCodec11SendRecvCodec
--- SKIP: TestCodec11SendRecvCodec (0.00 seconds)
codec_test.go💯 Test11SendRecvCodec norun
=== RUN TestConnBadVer10One
--- SKIP: TestConnBadVer10One (0.00 seconds)
connbv_test.go:30: TestConnBadVer10One no 1.0 only servers available
=== RUN TestConnBadVer10Two
--- SKIP: TestConnBadVer10Two (0.00 seconds)
connbv_test.go:50: TestConnBadVer10Two norun, set STOMP_TESTBV
=== RUN TestConnBadVer10Three
--- SKIP: TestConnBadVer10Three (0.00 seconds)
connbv_test.go:76: TestConnBadVer10Three norun, set STOMP_TESTBV
=== RUN TestConnDiscNetconn
=== RUN TestConnDiscStompConn
=== RUN TestConnDiscStompDisc
=== RUN TestConnDiscStompDiscReceipt
=== RUN TestConnBodyLen
=== RUN TestConn11p
=== RUN TestConn11Receipt
=== RUN TestEconBad
=== RUN TestSetProtocolLevel
=== RUN TestConnRespData
=== RUN TestDataFrameBasic
=== RUN TestDataMessageBasic
=== RUN TestDataprotocols
=== RUN TestDataProtocols
=== RUN TestDataError
=== RUN TestDataMessageSize
=== RUN TestBrokerCmdVal
=== RUN TestHB10
=== RUN TestHB11NoHeader
=== RUN TestHB11ZeroHeader
=== RUN TestHB11InitErrors
=== RUN TestHB11Connect
--- SKIP: TestHB11Connect (0.00 seconds)
hb_test.go:160: TestHB11Connect norun, need 1.1+
=== RUN TestHB11NoSend
--- SKIP: TestHB11NoSend (0.00 seconds)
hb_test.go:192: TestHB11NoSend norun, need 1.1+
=== RUN TestHB11NoReceive
--- SKIP: TestHB11NoReceive (0.00 seconds)
hb_test.go:236: TestHB11NoReceive norun, need 1.1+
=== RUN TestHB11SendReceive
--- SKIP: TestHB11SendReceive (0.00 seconds)
hb_test.go:276: TestHB11SendReceive norun, need 1.1+
=== RUN TestHB11SendReceiveApollo
--- SKIP: TestHB11SendReceiveApollo (0.00 seconds)
hb_test.go:323: TestHB11SendReceiveApollo norun, need 1.1+
=== RUN TestHB11SendReceiveApolloRev
--- SKIP: TestHB11SendReceiveApolloRev (0.00 seconds)
hb_test.go:371: TestHB11SendReceiveApolloRev norun, need 1.1+
=== RUN TestDataHeadersBasic
=== RUN TestDataHeadersUTF8
=== RUN TestDataHeadersClone
=== RUN TestDataHeadersAddDelete
=== RUN TestDataHeadersContainsKV
=== RUN TestDataHeadersCompare
=== RUN TestDataHeadersSize
=== RUN TestDataHeadersEmtKV
=== RUN TestLoggerBasic
=== RUN TestBytes0
=== RUN TestBytes1
=== RUN TestNilHeaders
=== RUN TestMax
=== RUN TestHasValue
=== RUN TestUuid
=== RUN TestBadHeaders
=== RUN TestNackErrors
=== RUN TestSendBasic
=== RUN TestSendMultiple
=== RUN TestSendBytesBasic
=== RUN TestSendBytesMultiple
=== RUN TestSubNoSub
=== RUN TestSubNoIdOnce
=== RUN TestSubNoIdTwice10
=== RUN TestSubNoIdTwice11p
--- SKIP: TestSubNoIdTwice11p (0.00 seconds)
sub_test.go:164: TestSubNoIdTwice11p norun, need 1.1+
=== RUN TestSubUnsubBasic
=== RUN TestSubUnsubBasic10
=== RUN TestSubEstablishSubscription
=== RUN TestSubSetCap
--- SKIP: TestSubSetCap (0.00 seconds)
sub_test.go:335: TestSubSetCap norun, need 1.1+
=== RUN TestTransErrors
=== RUN TestTransSend
=== RUN TestTransSendRollback
=== RUN TestTransMessageOrder
=== RUN TestUnsubNoHdr
=== RUN TestUnsubNoId
=== RUN TestUnsubBadId
=== RUN TestShovel11
--- SKIP: TestShovel11 (0.00 seconds)
wrsubrduns_test.go:29: Test11Shovel norun, need 1.1+
panic: runtime error: invalid memory address or nil pointer dereference
[signal 0xb code=0x1 addr=0x20 pc=0x4edf0c]

goroutine 153 [running]:
runtime.panic(0x5e9760, 0x730e13)
/usr/local/go/src/pkg/runtime/panic.c:279 +0xf5
bufio.(_Writer).flush(0xc208042c00, 0x0, 0x0)
/usr/local/go/src/pkg/bufio/bufio.go:530 +0xdc
bufio.(_Writer).Flush(0xc208042c00, 0x0, 0x0)
/usr/local/go/src/pkg/bufio/bufio.go:519 +0x39
github.com/danieltmiles/stompngo.(_Connection).wireWrite(0xc208054100, 0x60c090, 0x7, 0xc208042b80, 0x4, 0x4, 0x735b30, 0x0, 0x0, 0xc20807c000)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/writer.go:64 +0x38e
github.com/danieltmiles/stompngo.(_Connection).writer(0xc208054100)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/writer.go:35 +0xcb
created by github.com/danieltmiles/stompngo.Connect
/home/dmiles/go/src/github.com/danieltmiles/stompngo/connect.go:93 +0x656

goroutine 16 [chan receive]:
testing.RunTests(0x675460, 0x733520, 0x49, 0x49, 0x1)
/usr/local/go/src/pkg/testing/testing.go:525 +0x519
testing.Main(0x675460, 0x733520, 0x49, 0x49, 0x72f160, 0x4, 0x4, 0x7399c0, 0x0, 0x0)
/usr/local/go/src/pkg/testing/testing.go:435 +0x84
main.main()
github.com/danieltmiles/stompngo/_test/_testmain.go:199 +0x9c

goroutine 19 [finalizer wait]:
runtime.park(0x4137f0, 0x7356c0, 0x734229)
/usr/local/go/src/pkg/runtime/proc.c:1369 +0x89
runtime.parkunlock(0x7356c0, 0x734229)
/usr/local/go/src/pkg/runtime/proc.c:1385 +0x3b
runfinq()
/usr/local/go/src/pkg/runtime/mgc0.c:2644 +0xcf
runtime.goexit()
/usr/local/go/src/pkg/runtime/proc.c:1445

goroutine 20 [runnable]:
github.com/danieltmiles/stompngo.Connect(0x0, 0x0, 0x72ed80, 0x4, 0x4, 0x4, 0x0, 0x0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/connect.go:96 +0x75c
github.com/danieltmiles/stompngo.TestAckErrors(0xc208050120)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/ack_test.go:72 +0xdf
testing.tRunner(0xc208050120, 0x733520)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 21 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 22 [chan receive]:
testing.(*T).Parallel(0xc2080501b0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestAckSameConn(0xc2080501b0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/ack_test.go:98 +0x35
testing.tRunner(0xc2080501b0, 0x733538)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 23 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 24 [chan receive]:
testing.(*T).Parallel(0xc208050240)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestAckDiffConn(0xc208050240)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/ack_test.go:179 +0x35
testing.tRunner(0xc208050240, 0x733550)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 25 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 26 [chan receive]:
testing.(*T).Parallel(0xc2080502d0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestCodecEncodeBasic(0xc2080502d0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/codec_test.go:52 +0x35
testing.tRunner(0xc2080502d0, 0x733568)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 27 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 28 [chan receive]:
testing.(*T).Parallel(0xc208050360)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestCodecDecodeBasic(0xc208050360)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/codec_test.go:68 +0x35
testing.tRunner(0xc208050360, 0x733580)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 29 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 34 [chan receive]:
testing.(*T).Parallel(0xc208050630)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestConnDiscNetconn(0xc208050630)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/conndisc_test.go:43 +0x27
testing.tRunner(0xc208050630, 0x7335f8)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 35 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 36 [chan receive]:
testing.(*T).Parallel(0xc2080506c0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestConnDiscStompConn(0xc2080506c0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/conndisc_test.go:52 +0x35
testing.tRunner(0xc2080506c0, 0x733610)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 37 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 38 [chan receive]:
testing.(*T).Parallel(0xc208050750)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestConnDiscStompDisc(0xc208050750)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/conndisc_test.go:117 +0x32
testing.tRunner(0xc208050750, 0x733628)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 39 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 40 [chan receive]:
testing.(*T).Parallel(0xc2080507e0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestConnDiscStompDiscReceipt(0xc2080507e0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/conndisc_test.go:132 +0x35
testing.tRunner(0xc2080507e0, 0x733640)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 41 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 42 [chan receive]:
testing.(*T).Parallel(0xc208050870)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestConnBodyLen(0xc208050870)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/conndisc_test.go:160 +0x32
testing.tRunner(0xc208050870, 0x733658)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 43 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 44 [chan receive]:
testing.(*T).Parallel(0xc208050000)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestConn11p(0xc208050000)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/conndisc_test.go:179 +0x35
testing.tRunner(0xc208050000, 0x733670)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 45 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 46 [chan receive]:
testing.(*T).Parallel(0xc2080503f0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestConn11Receipt(0xc2080503f0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/conndisc_test.go:211 +0x35
testing.tRunner(0xc2080503f0, 0x733688)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 47 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 48 [chan receive]:
testing.(*T).Parallel(0xc208050480)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestEconBad(0xc208050480)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/conndisc_test.go:229 +0x32
testing.tRunner(0xc208050480, 0x7336a0)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 49 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 50 [chan receive]:
testing.(*T).Parallel(0xc208050510)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestSetProtocolLevel(0xc208050510)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/conndisc_test.go:278 +0x35
testing.tRunner(0xc208050510, 0x7336b8)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 51 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 52 [chan receive]:
testing.(*T).Parallel(0xc208050900)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestConnRespData(0xc208050900)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/conndisc_test.go:300 +0x35
testing.tRunner(0xc208050900, 0x7336d0)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 53 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 54 [chan receive]:
testing.(*T).Parallel(0xc208050990)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataFrameBasic(0xc208050990)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/data_test.go:41 +0x35
testing.tRunner(0xc208050990, 0x7336e8)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 55 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 56 [chan receive]:
testing.(*T).Parallel(0xc208050a20)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataMessageBasic(0xc208050a20)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/data_test.go:62 +0x35
testing.tRunner(0xc208050a20, 0x733700)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 57 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 58 [chan receive]:
testing.(*T).Parallel(0xc208050ab0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataprotocols(0xc208050ab0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/data_test.go:83 +0x35
testing.tRunner(0xc208050ab0, 0x733718)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 59 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 60 [chan receive]:
testing.(*T).Parallel(0xc208050b40)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataProtocols(0xc208050b40)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/data_test.go:113 +0x35
testing.tRunner(0xc208050b40, 0x733730)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 61 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 62 [chan receive]:
testing.(*T).Parallel(0xc208050bd0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataError(0xc208050bd0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/data_test.go:126 +0x32
testing.tRunner(0xc208050bd0, 0x733748)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 63 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 64 [chan receive]:
testing.(*T).Parallel(0xc208050c60)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataMessageSize(0xc208050c60)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/data_test.go:138 +0x35
testing.tRunner(0xc208050c60, 0x733760)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 65 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 66 [chan receive]:
testing.(*T).Parallel(0xc208050cf0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestBrokerCmdVal(0xc208050cf0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/data_test.go:156 +0x35
testing.tRunner(0xc208050cf0, 0x733778)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 67 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 68 [chan receive]:
testing.(*T).Parallel(0xc208050d80)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestHB10(0xc208050d80)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/hb_test.go:31 +0x27
testing.tRunner(0xc208050d80, 0x733790)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 69 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 70 [chan receive]:
testing.(*T).Parallel(0xc208050e10)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestHB11NoHeader(0xc208050e10)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/hb_test.go:47 +0x32
testing.tRunner(0xc208050e10, 0x7337a8)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 71 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 72 [chan receive]:
testing.(*T).Parallel(0xc208050ea0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestHB11ZeroHeader(0xc208050ea0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/hb_test.go:68 +0x35
testing.tRunner(0xc208050ea0, 0x7337c0)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 73 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 74 [chan receive]:
testing.(*T).Parallel(0xc208050f30)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestHB11InitErrors(0xc208050f30)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/hb_test.go:89 +0x35
testing.tRunner(0xc208050f30, 0x7337d8)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 75 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 82 [chan receive]:
testing.(*T).Parallel(0xc208051320)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataHeadersBasic(0xc208051320)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/headers_test.go:27 +0x35
testing.tRunner(0xc208051320, 0x733880)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 83 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 84 [chan receive]:
testing.(*T).Parallel(0xc2080513b0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataHeadersUTF8(0xc2080513b0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/headers_test.go:59 +0x35
testing.tRunner(0xc2080513b0, 0x733898)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 85 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 86 [chan receive]:
testing.(*T).Parallel(0xc208051440)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataHeadersClone(0xc208051440)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/headers_test.go:93 +0x35
testing.tRunner(0xc208051440, 0x7338b0)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 87 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 88 [chan receive]:
testing.(*T).Parallel(0xc2080514d0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataHeadersAddDelete(0xc2080514d0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/headers_test.go:105 +0x35
testing.tRunner(0xc2080514d0, 0x7338c8)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 89 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 90 [chan receive]:
testing.(*T).Parallel(0xc208051560)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataHeadersContainsKV(0xc208051560)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/headers_test.go:127 +0x32
testing.tRunner(0xc208051560, 0x7338e0)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 91 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 92 [chan receive]:
testing.(*T).Parallel(0xc2080515f0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataHeadersCompare(0xc2080515f0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/headers_test.go:143 +0x35
testing.tRunner(0xc2080515f0, 0x7338f8)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 93 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 94 [chan receive]:
testing.(*T).Parallel(0xc208051680)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataHeadersSize(0xc208051680)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/headers_test.go:170 +0x35
testing.tRunner(0xc208051680, 0x733910)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 95 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 96 [chan receive]:
testing.(*T).Parallel(0xc208051710)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestDataHeadersEmtKV(0xc208051710)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/headers_test.go:190 +0x35
testing.tRunner(0xc208051710, 0x733928)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 97 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 98 [chan receive]:
testing.(*T).Parallel(0xc2080517a0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestLoggerBasic(0xc2080517a0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/logger_test.go:29 +0x32
testing.tRunner(0xc2080517a0, 0x733940)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 99 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 100 [chan receive]:
testing.(*T).Parallel(0xc208051830)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestBytes0(0xc208051830)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/misc_test.go:27 +0x35
testing.tRunner(0xc208051830, 0x733958)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 101 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 102 [chan receive]:
testing.(*T).Parallel(0xc2080518c0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestBytes1(0xc2080518c0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/misc_test.go:79 +0x35
testing.tRunner(0xc2080518c0, 0x733970)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 103 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 104 [chan receive]:
testing.(*T).Parallel(0xc208051950)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestNilHeaders(0xc208051950)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/misc_test.go:131 +0x35
testing.tRunner(0xc208051950, 0x733988)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 105 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 106 [chan receive]:
testing.(*T).Parallel(0xc2080519e0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestMax(0xc2080519e0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/misc_test.go:197 +0x32
testing.tRunner(0xc2080519e0, 0x7339a0)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 107 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 108 [chan receive]:
testing.(*T).Parallel(0xc208051a70)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestHasValue(0xc208051a70)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/misc_test.go:214 +0x32
testing.tRunner(0xc208051a70, 0x7339b8)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 109 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 110 [chan receive]:
testing.(*T).Parallel(0xc208051b00)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestUuid(0xc208051b00)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/misc_test.go:228 +0x32
testing.tRunner(0xc208051b00, 0x7339d0)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 111 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 112 [chan receive]:
testing.(*T).Parallel(0xc208051b90)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestBadHeaders(0xc208051b90)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/misc_test.go:242 +0x35
testing.tRunner(0xc208051b90, 0x7339e8)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 113 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 114 [chan receive]:
testing.(*T).Parallel(0xc208051c20)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestNackErrors(0xc208051c20)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/nack_test.go:66 +0x35
testing.tRunner(0xc208051c20, 0x733a00)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 115 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 116 [chan receive]:
testing.(*T).Parallel(0xc208051cb0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestSendBasic(0xc208051cb0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/send_test.go:28 +0x35
testing.tRunner(0xc208051cb0, 0x733a18)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 117 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 118 [chan receive]:
testing.(*T).Parallel(0xc208051d40)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestSendMultiple(0xc208051d40)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/send_test.go:60 +0x32
testing.tRunner(0xc208051d40, 0x733a30)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 119 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 120 [chan receive]:
testing.(*T).Parallel(0xc208051dd0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestSendBytesBasic(0xc208051dd0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/sendbytes_test.go:28 +0x35
testing.tRunner(0xc208051dd0, 0x733a48)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 121 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 122 [chan receive]:
testing.(*T).Parallel(0xc208051e60)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestSendBytesMultiple(0xc208051e60)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/sendbytes_test.go:60 +0x32
testing.tRunner(0xc208051e60, 0x733a60)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 123 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 124 [chan receive]:
testing.(*T).Parallel(0xc208051ef0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestSubNoSub(0xc208051ef0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/sub_test.go:31 +0x35
testing.tRunner(0xc208051ef0, 0x733a78)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 125 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 126 [chan receive]:
testing.(*T).Parallel(0xc208064000)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestSubNoIdOnce(0xc208064000)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/sub_test.go:56 +0x35
testing.tRunner(0xc208064000, 0x733a90)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 127 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 128 [chan receive]:
testing.(*T).Parallel(0xc208064090)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestSubNoIdTwice10(0xc208064090)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/sub_test.go:90 +0x14c
testing.tRunner(0xc208064090, 0x733aa8)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 129 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 131 [chan receive]:
testing.(*T).Parallel(0xc2080641b0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestSubUnsubBasic(0xc2080641b0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/sub_test.go:215 +0x35
testing.tRunner(0xc2080641b0, 0x733ad8)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 132 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 133 [chan receive]:
testing.(*T).Parallel(0xc208064240)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestSubUnsubBasic10(0xc208064240)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/sub_test.go:266 +0x12f
testing.tRunner(0xc208064240, 0x733af0)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 134 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 135 [chan receive]:
testing.(*T).Parallel(0xc2080642d0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestSubEstablishSubscription(0xc2080642d0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/sub_test.go:309 +0x32
testing.tRunner(0xc2080642d0, 0x733b08)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 136 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 138 [chan receive]:
testing.(*T).Parallel(0xc2080643f0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestTransErrors(0xc2080643f0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/trans_test.go:28 +0x35
testing.tRunner(0xc2080643f0, 0x733b38)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 139 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 140 [chan receive]:
testing.(*T).Parallel(0xc208064480)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestTransSend(0xc208064480)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/trans_test.go:119 +0x35
testing.tRunner(0xc208064480, 0x733b50)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 141 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 142 [chan receive]:
testing.(*T).Parallel(0xc208064510)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestTransSendRollback(0xc208064510)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/trans_test.go:168 +0x35
testing.tRunner(0xc208064510, 0x733b68)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 143 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 144 [chan receive]:
testing.(*T).Parallel(0xc2080645a0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestTransMessageOrder(0xc2080645a0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/trans_test.go:235 +0x35
testing.tRunner(0xc2080645a0, 0x733b80)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 145 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 146 [chan receive]:
testing.(*T).Parallel(0xc208064630)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestUnsubNoHdr(0xc208064630)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/unsub_test.go:49 +0x35
testing.tRunner(0xc208064630, 0x733b98)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 147 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 148 [chan receive]:
testing.(*T).Parallel(0xc2080646c0)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestUnsubNoId(0xc2080646c0)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/unsub_test.go:77 +0x35
testing.tRunner(0xc2080646c0, 0x733bb0)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 149 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 150 [chan receive]:
testing.(*T).Parallel(0xc208064750)
/usr/local/go/src/pkg/testing/testing.go:388 +0x76
github.com/danieltmiles/stompngo.TestUnsubBadId(0xc208064750)
/home/dmiles/go/src/github.com/danieltmiles/stompngo/unsub_test.go:104 +0x35
testing.tRunner(0xc208064750, 0x733bc8)
/usr/local/go/src/pkg/testing/testing.go:422 +0x8b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:504 +0x8db

goroutine 151 [chan receive]:
testing.func·007()
/usr/local/go/src/pkg/testing/testing.go:508 +0x5b
created by testing.RunTests
/usr/local/go/src/pkg/testing/testing.go:509 +0x999

goroutine 145 [syscall]:
runtime.goexit()
/usr/local/go/src/pkg/runtime/proc.c:1445
exit status 2
FAIL github.com/danieltmiles/stompngo 0.039s

Can this project be used to dock ActiveMQ in a production environment with 100,000 throughput?

when i send via Stompngo I seem to be publishing messages as ActiveMqByte Messages

Jan 06 15:26:32 DEBUG tropo-cdr2 [voiceCdrSpringListenerBillwise-1] BillingBridge - Received message of type [class org.apache.activemq.command.ActiveMQBytesMessage] from consumer [ActiveMQMessageConsumer { value=ID:tropo-cdr2.prod.us-west-2xxxxxxx-43243-1417638886581-0:0:1:1, started=true }] of session [PooledSession { ActiveMQSession {id=ID:tropo-cdr2.prod.us-west-2.aws.xxxxxx-43243-1417638886581-0:0:1,started=true} }]

I am using the Send function which as I understand it should be sending a string to the queue

package main

import (
    "bufio"
    "fmt"
    "github.com/gmallard/stompngo"
    "gopkg.in/alecthomas/kingpin.v1"
    "net"
    "os"
)

var (
    queueName     = kingpin.Flag("queue", "Destination").Default("/queue/client_test").Short('q').String()
    serverAddr    = kingpin.Flag("server", "STOMP server endpoint").Default("127.0.0.1").Short('s').String()
    serverPort    = kingpin.Flag("port", "STOMP server port").Default("61613").Short('p').String()
    fileToProcess = kingpin.Flag("file", "File to process").Short('f').String()
    workerCount   = kingpin.Flag("workers", "Number of workers to send/receive").Short('w').Int()
    serverUser    = kingpin.Flag("user", "Username").OverrideDefaultFromEnvar("STOMP_USER").String()
    serverPass    = kingpin.Flag("pass", "Password").OverrideDefaultFromEnvar("STOMP_PASS").String()

    client Client
    done   = make(chan bool)
)

//var done = make(chan bool)

type Client struct {
    Host     string
    Port     string
    User     string
    Password string
    Uuid     string
    Queue    string

    NetConnection   net.Conn
    StompConnection *stompngo.Connection
}

func init() {
    kingpin.Version("0.0.1")
    kingpin.Parse()

    // Set default of 4 workers
    if *workerCount == 0 {
        *workerCount = 4
    }

}

// Setups connection options
func (client *Client) setOpts() {

    client.Host = *serverAddr
    client.Port = *serverPort
    client.Uuid = stompngo.Uuid()
    client.Queue = *queueName

    if *serverUser != "" {
        client.User = *serverUser
    }

    if *serverPass != "" {
        client.Password = *serverPass
    }
}

// Creates net connection
func (client *Client) netConnection() (conn net.Conn, err error) {
    conn, err = net.Dial("tcp", net.JoinHostPort(client.Host, client.Port))
    if err != nil {
        fmt.Println(err)
        return nil, err
    }

    client.NetConnection = conn
    return
}

func (client *Client) stompConnection() *stompngo.Connection {
    headers := stompngo.Headers{
        "accept-version", "1.1",
        "host", client.Host,
        "login", client.User,
        "passcode", client.Password,
    }

    conn, err := stompngo.Connect(client.NetConnection, headers)
    if err != nil {
        fmt.Println(err)
        os.Exit(0)
    }

    client.StompConnection = conn
    return conn
}

func (client *Client) Connect() (conn *stompngo.Connection) {
    client.setOpts()
    client.netConnection()

    conn = client.stompConnection()

    return
}

func (client *Client) Disconnect() {
    client.StompConnection.Disconnect(stompngo.Headers{})
    client.NetConnection.Close()
}

//  Start main
//
//
func main() {

    fmt.Println("Starting connection")
    _ = client.Connect()
    defer client.Disconnect()

    dataCh := make(chan string, *workerCount)

    // Start workers
    fmt.Println("Create workers")
    for id := 1; id <= *workerCount; id++ {
        go sender(id, &client, dataCh)
    }

    //  Start reader go routine
    fmt.Println("Read file")
    go fileReader(*fileToProcess, dataCh)

    <-done

    fmt.Println("Done")
}

//  Read from file and put data line by line on channel
func fileReader(path string, dataCh chan<- string) {
    inFile, _ := os.Open(path)
    defer inFile.Close()
    scanner := bufio.NewScanner(inFile)
    scanner.Split(bufio.ScanLines)

    for scanner.Scan() {
        dataCh <- scanner.Text()
    }
    close(dataCh)
}

//  Read from channel and put on queue
func sender(id int, client *Client, dataCh <-chan string) {

    for message := range dataCh {
        headers := stompngo.Headers{"destination", *queueName, "id", client.Uuid, "persistent", "true"}
        err := client.StompConnection.Send(headers, message)
        if err != nil {
            fmt.Println(err)
        }
    }

    done <- true
}

@gmallard

Recently, we found a blocking case with calling Connection.Disconnect(). It hang on

I suppose it is due to the connection status of stompngo.Connection is based on Connection.connected, which may encounter race condition.

In stompngo/writer.go:

func (c *Connection) writer() {
writerLoop:
	for {
		select {
		case d := <-c.output:
			c.log("WTR_WIREWRITE start")
			c.wireWrite(d)
			logLock.Lock()
			if c.logger != nil {
				c.logx("WTR_WIREWRITE COMPLETE", d.frame.Command, d.frame.Headers,
					HexData(d.frame.Body))
			}
			logLock.Unlock()
			if d.frame.Command == DISCONNECT {
				break writerLoop // we are done with this connection
			}
		case _ = <-c.ssdc:
			c.log("WTR_WIREWRITE shutdown S received")
			break writerLoop
		case _ = <-c.wtrsdc:
			c.log("WTR_WIREWRITE shutdown W received")
			break writerLoop
		}
	} // of for
	//
	c.connected = false
	c.log("WTR_SHUTDOWN", time.Now())
}

When writer receive signal from c.wtrsdc, before it break the writerLoop. It has chance for a Disconnect call bypass c.connected check.

Possible solution is using c.connected as a chan instead of bool. use close(c.connected) instead of set c.connected to false.

So, following line

can be replaced by

select {
    case c.output <- wiredata{f, r}:
    case <- c.connected:
        //Already disconnected
        return
}

What do you think of it?

Hello,
Thanks a lot for developing this nice tool.
I've got a question regarding the connection Disconnect function; when I use it, it hangs, and when I debug it, the debugger stops at the following line:

Hi , I have read the source code of stompngo and find something hard to understand.

Here's a reader go routine which can be set read timeout by conn.ReadDealLine

func (c *Connection) setReadDeadline() {
	if c.dld.rde && c.dld.rds {
		_ = c.netconn.SetReadDeadline(time.Now().Add(c.dld.rdld))
	}
}

The reader use c.dld.rde && c.dld.rds to check if read deadline will be set.
Buf a writer use c.dld.wde && c.dld.dns to check if write deadline will be set.
Should it use c.dld.wde && c.dld.wds instead ?

func (c *Connection) wireWrite(d wiredata) {
	f := &d.frame
	// fmt.Printf("WWD01 f:[%v]\n", f)
	switch f.Command {
	case "\n": // HeartBeat frame
		if c.dld.wde && c.dld.dns {
			_ = c.netconn.SetWriteDeadline(time.Now().Add(c.dld.wdld))
		}

Looking forward to your reply

AMQ 5.6.0 is implemented like Rabbit: it does not return duplicate headers.

The test needs to handle AMQ as is currently done for Rabbit testing.

For protocol levels 1.1 and 1.2 a destination header is not required.

The 1.0 level requires either destination or id headers.

The current implementation erroneously requires the destination header for 1.1 and 1.2.

It would be great if there was a way to connect via SSL. We're using stomp+ssl over port 61613, not regular tcp so unfortunately we can't use this at the moment.

I don't know why this occurred, but I got this error during one of my test runs:

panic: non-positive interval for NewTicker
time.NewTicker
    go/src/time/tick.go:32
github.com/gmallard/stompngo.(*Connection).receiveTicker
    src/github.com/gmallard/stompngo/heartbeats.go:163
github.com/gmallard/stompngo.(*Connection).initializeHeartBeats
    src/github.com/gmallard/stompngo/heartbeats.go:183

It only happened to me once, so it's probably something transient. Just wanted to report it as this seems like something that would be very simple to check for so that it's prevented in the future.

Not sure when this started. Needs quite a bit more analysis.

stompngo at 048c068

Apologies if this has already been suggested or if this is not the proper channel for suggesting feature requests. Also, I'm totally willing to make this PR myself, I just want to see how the authors feel about it before actually going about making the code changes.

Abstract stompngo.Connection to an interface to allow mock generators to mock the connection and allow for easier testing. Use a different name to not break backwards compatibility. So something like:

type STOMPConnection interface {
	Send(stompngo.Headers, string) error
	Connected() bool
	Disconnect(headers stompngo.Headers) error
	Subscribe(headers stompngo.Headers) (<-chan stompngo.MessageData, error)
	Unsubscribe(headers stompngo.Headers) error
	Ack(headers stompngo.Headers) error
	Nack(headers stompngo.Headers) error
}

There are two time.Ticker used in heartbeats which are not Stop() after exit running.
Resources will be slowly exhausted.

Using the CONNECT Headers:

stompngo.Headers{"accept-version", "1.1", ......}

And:

• Connect is made to a 1.0 broker
• The broker (correctly) returns no 1.1 related headers
• The connection succeeds, and shows a protocol of "1.0"

The client specifically excluded "1.0", and this connection should fail.

The first attempt to send after a broker drops the connection returns no [expected] error. Subsequent attempts to send will return an expected error.

Hello,

I came across this debugging an app I'm working on. I can see that the app is gettng an error (EOF), which is causing the reader and writer to shut down. However, the system is still trying to send heartbeats afterwards. Unfortunately, I tried to trace through this a bit to no avail. The issue with this is that it looks like the client doesn't die and continues to send heartbeats. My preference would be that if the connection dies, the client sends back an error.

I'll keep a copy of this log file handy in case you need more info. Unfortunately, I can't share it.

The entire package needs to be tested with -race. The go race detector did not exist when this package was first written and published.

Current operation:

When a client inadvertantly:

1. obtains a net.Conn implementation using net.Dial (not tls.Dial) and
2. connects to a broker port configured for SSL

The package currently returns EBADFRM (bad frame).

Suggested operation:

The Error returned should be more specific.

It is possible to tell that the reply from the broker is SSL "handshake data" (the Ruby stomp gem does this).

The package should be changed to:

• detect handshake data being returned
• return a more descriptive Error the the client

Hand shake data will look like:

"\x15\x03\x03\x00"

The Error should be something like:

Error("probable bad SSL configuration")

Specification available:

http://stomp.github.com/

The Subscribe implementation is missing standard calls to c.log at start and end of the method.

TODO: check other connection methods as well.

Hello,

We ran across an issue with RabbitMQ and stompngo where if we spammed RabbitMQ fast enough, we would get Heartbeat errors. While debugging this, it appears that it got into a state where the connection crapped out and it the bufio writer would not write the data. Put a timeout deadline on network connection before sending data helped a bit, but it still locked up.

Regardless of what happens, it may be a good idea to allow the user to specify a timeout for read/write operations. This way, if there any network issues, they aren't silently ignored in the worst case.

I just noticed this doing some load testing.

Note: this does not happen with GOMAXPROCS defaulted.

When:

a) GOMAXPROCS is set to NumCPUs
b) Many network connections (> 60 ?) are in use

Initial analysis of experiments to date indicates this is a result of logic in a5973f2