sub_test.go

   1 //
   2 // Copyright © 2011-2016 Guy M. Allard
   3 //
   4 // Licensed under the Apache License, Version 2.0 (the "License");
   5 // you may not use this file except in compliance with the License.
   6 // You may obtain a copy of the License at
   7 //
   8 //     http://www.apache.org/licenses/LICENSE-2.0
   9 //
  10 // Unless required by applicable law or agreed to in writing, software
  11 // distributed under the License is distributed on an "AS IS" BASIS,
  12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 // See the License for the specific language governing permissions and
  14 // limitations under the License.
  15 //
  16
  17 package stompngo
  18
  19 import (
  20         "log"
  21         "os"
  22         "testing"
  23         "time"
  24 )
  25
  26 /*
  27         Test Subscribe, no destination.
  28 */
  29 func TestSubNoDest(t *testing.T) {
  30
  31         n, _ := openConn(t)
  32         ch := check11(TEST_HEADERS)
  33         conn, _ := Connect(n, ch)
  34         // Subscribe, no dest
  35         _, e := conn.Subscribe(empty_headers)
  36         if e == nil {
  37                 t.Fatalf("Expected subscribe error, got [nil]\n")
  38         }
  39         if e != EREQDSTSUB {
  40                 t.Fatalf("Subscribe error, expected [%v], got [%v]\n", EREQDSTSUB, e)
  41         }
  42         //
  43         _ = conn.Disconnect(empty_headers)
  44         _ = closeConn(t, n)
  45 }
  46
  47 /*
  48         Test subscribe, no ID.
  49 */
  50 func TestSubNoIdOnce(t *testing.T) {
  51
  52         n, _ := openConn(t)
  53         ch := check11(TEST_HEADERS)
  54         conn, _ := Connect(n, ch)
  55         //
  56         d := tdest("/queue/subunsub.genl.01")
  57         sbh := Headers{HK_DESTINATION, d}
  58         //
  59         s, e := conn.Subscribe(sbh)
  60         if e != nil {
  61                 t.Fatalf("Expected no subscribe error, got [%v]\n", e)
  62         }
  63         if s == nil {
  64                 t.Fatalf("Expected subscribe channel, got [nil]\n")
  65         }
  66         select {
  67         case v := <-conn.MessageData:
  68                 t.Fatalf("Unexpected frame received, got [%v]\n", v)
  69         default:
  70         }
  71         //
  72         _ = conn.Disconnect(empty_headers)
  73         _ = closeConn(t, n)
  74 }
  75
  76 /*
  77         Test subscribe, no ID, twice to same destination, protocol level 1.0.
  78 */
  79 func TestSubNoIdTwice10(t *testing.T) {
  80         if os.Getenv("STOMP_TEST11p") != "" {
  81                 t.Skip("TestSubNoIdTwice10 norun, need 1.0")
  82         }
  83
  84         t.Log("TestSubNoIdTwice10", "starts")
  85         n, _ := openConn(t)
  86         ch := check11(TEST_HEADERS)
  87         conn, _ := Connect(n, ch)
  88         //l := log.New(os.Stdout, "", log.Ldate|log.Lmicroseconds)
  89         //conn.SetLogger(l)
  90         //
  91         if conn.Protocol() != SPL_10 {
  92                 t.Fatalf("Protocol error, got [%v], expected [%v]\n", conn.Protocol(), SPL_10)
  93         }
  94         //
  95         d := tdest("/queue/subdup.p10.01")
  96         sbh := Headers{HK_DESTINATION, d}
  97         // First time
  98         sc, e := conn.Subscribe(sbh)
  99         if e != nil {
 100                 t.Fatalf("Expected no subscribe error (T1), got [%v]\n", e)
 101         }
 102         if sc == nil {
 103                 t.Fatalf("Expected subscribe channel (T1), got [nil]\n")
 104         }
 105         time.Sleep(500 * time.Millisecond) // give a broker a break
 106         select {
 107         case v := <-sc:
 108                 t.Fatalf("Unexpected frame received (T1), got [%v]\n", v)
 109         case v := <-conn.MessageData:
 110                 t.Fatalf("Unexpected frame received (T1), got [%v]\n", v)
 111         default:
 112         }
 113         // Second time
 114         sc, e = conn.Subscribe(sbh)
 115         if e == EDUPSID {
 116                 t.Fatalf("Expected no subscribe error (T2), got [%v]\n", e)
 117         }
 118         if e != nil {
 119                 t.Fatalf("Expected no subscribe error (T2), got [%v]\n", e)
 120         }
 121         if sc == nil {
 122                 t.Fatalf("Expected subscribe channel (T2), got nil\n")
 123         }
 124         time.Sleep(500 * time.Millisecond) // give a broker a break
 125         // Stomp 1.0 brokers are allowed significant latitude regarding a response
 126         // to a duplicate subscription request.  Currently, only do these checks for
 127         // brokers other than AMQ.  AMQ does not return an ERROR frame for duplicate
 128         // subscriptions with 1.0, choosing to ignore it.
 129         // Apollo and RabbitMQ both return an ERROR frame *and* tear down the
 130         // connection.
 131         if os.Getenv("STOMP_APOLLO") != "" || os.Getenv("STOMP_RMQ") != "" {
 132                 // fmt.Println("sccheck runs ....", conn.Connected())
 133                 select {
 134                 case v := <-sc:
 135                         t.Logf("Server frame expected and received (T2-A), got [%v] [%v] [%v] [%s]\n",
 136                                 v.Message.Command, v.Error, v.Message.Headers, string(v.Message.Body))
 137                 case v := <-conn.MessageData:
 138                         t.Logf("Server frame expected and received (T2-B), got [%v] [%v] [%v] [%s]\n",
 139                                 v.Message.Command, v.Error, v.Message.Headers, string(v.Message.Body))
 140                 default:
 141                         t.Fatalf("Server frame expected (T2-E), not received.\n")
 142                 }
 143         }
 144         // For both Apollo and RabbitMQ, the connection teardown by the server can
 145         // mean the client side connection is no longer usable.
 146         if os.Getenv("STOMP_APOLLO") == "" && os.Getenv("STOMP_RMQ") == "" {
 147                 _ = conn.Disconnect(empty_headers)
 148                 _ = closeConn(t, n)
 149         }
 150         t.Log("TestSubNoIdTwice10", "ends")
 151 }
 152
 153 /*
 154         Test subscribe, no ID, twice to same destination, protocol level 1.1+.
 155 */
 156 func TestSubNoIdTwice11p(t *testing.T) {
 157         if os.Getenv("STOMP_TEST11p") == "" {
 158                 t.Skip("TestSubNoIdTwice11p norun, need 1.1+")
 159         }
 160
 161         t.Log("TestSubNoIdTwice11p", "starts")
 162         n, _ := openConn(t)
 163         ch := check11(TEST_HEADERS)
 164         conn, _ := Connect(n, ch)
 165         l := log.New(os.Stdout, "TSNI211P ", log.Ldate|log.Lmicroseconds)
 166         conn.SetLogger(l)
 167
 168         d := tdest("/queue/subdup.p11.01")
 169         id := "TestSubNoIdTwice11p"
 170         sbh := Headers{HK_DESTINATION, d, HK_ID, id}
 171         // First time
 172         t.Logf("%s\n", "INFO TestSubNoIdTwice11p - start 1st SUBSCRIBE")
 173         sc, e := conn.Subscribe(sbh)
 174         t.Logf("%s\n", "INFO TestSubNoIdTwice11p - end 1st SUBSCRIBE")
 175         if e != nil {
 176                 t.Fatalf("ERROR Expected no subscribe error (T1), got [%v]\n", e)
 177         }
 178         if sc == nil {
 179                 t.Fatalf("ERROR Expected subscribe channel (T2), got [nil]\n")
 180         }
 181         time.Sleep(500 * time.Millisecond) // give a broker a break
 182         select {
 183         case v := <-sc:
 184                 t.Fatalf("ERROR Unexpected frame received (T3), got [%v]\n", v)
 185         case v := <-conn.MessageData:
 186                 t.Fatalf("ERROR Unexpected frame received (T4), got [%v]\n", v)
 187         default:
 188         }
 189
 190         // Second time.  The stompngo package maintains a list of all current
 191         // subscription ids.  An attempt to subscribe using an existing id is
 192         // immediately rejected by the package (never actually sent to the broker).
 193         t.Logf("%s\n", "INFO TestSubNoIdTwice11p - start 2nd SUBSCRIBE")
 194         sc, e = conn.Subscribe(sbh)
 195         t.Logf("%s\n", "INFO TestSubNoIdTwice11p - end 2nd SUBSCRIBE")
 196         if e == nil {
 197                 t.Fatalf("ERROR Expected subscribe error (T5), got [nil]\n")
 198         }
 199         if e != EDUPSID {
 200                 t.Fatalf("ERROR Expected subscribe error (T6), [%v] got [%v]\n", EDUPSID, e)
 201         } else {
 202                 t.Logf("INFO wanted/got actual (T7), [%v]\n", e)
 203         }
 204         if sc != nil {
 205                 t.Fatalf("ERROR Expected nil subscribe channel (T8), got [%v]\n", sc)
 206         }
 207         _ = conn.Disconnect(empty_headers)
 208         _ = closeConn(t, n)
 209         t.Log("TestSubNoIdTwice11p", "ends")
 210 }
 211
 212 /*
 213         Test send, subscribe, read, unsubscribe.
 214 */
 215 func TestSubUnsubBasic(t *testing.T) {
 216
 217         n, _ := openConn(t)
 218         ch := check11(TEST_HEADERS)
 219         conn, _ := Connect(n, ch)
 220         //
 221         m := "A message"
 222         d := tdest("/queue/subunsub.basiconn.01")
 223         h := Headers{HK_DESTINATION, d}
 224         _ = conn.Send(h, m)
 225         //
 226         h = h.Add(HK_ID, d)
 227         sc, e := conn.Subscribe(h)
 228         if e != nil {
 229                 t.Fatalf("Expected no subscribe error, got [%v]\n", e)
 230         }
 231         if sc == nil {
 232                 t.Fatalf("Expected subscribe channel, got [nil]\n")
 233         }
 234
 235         // Read MessageData
 236         var md MessageData
 237         select {
 238         case md = <-sc:
 239         case md = <-conn.MessageData:
 240                 t.Fatalf("read channel error:  expected [nil], got: [%v]\n",
 241                         md.Message.Command)
 242         }
 243
 244         //
 245         if md.Error != nil {
 246                 t.Fatalf("Expected no message data error, got [%v]\n", md.Error)
 247         }
 248         mdm := md.Message
 249         rd := mdm.Headers.Value(HK_DESTINATION)
 250         if rd != d {
 251                 t.Fatalf("Expected destination [%v], got [%v]\n", d, rd)
 252         }
 253         ri := mdm.Headers.Value(HK_SUBSCRIPTION)
 254         if ri != d {
 255                 t.Fatalf("Expected subscription [%v], got [%v]\n", d, ri)
 256         }
 257         //
 258         e = conn.Unsubscribe(h)
 259         if e != nil {
 260                 t.Fatalf("Expected no unsubscribe error, got [%v]\n", e)
 261         }
 262         //
 263         _ = conn.Disconnect(empty_headers)
 264         _ = closeConn(t, n)
 265 }
 266
 267 /*
 268         Test send, subscribe, read, unsubscribe, 1.0 only, no sub id.
 269 */
 270 func TestSubUnsubBasic10(t *testing.T) {
 271         if os.Getenv("STOMP_TEST11p") != "" {
 272                 t.Skip("TestSubUnsubBasic10 norun, need 1.0")
 273         }
 274
 275         n, _ := openConn(t)
 276         ch := check11(TEST_HEADERS)
 277         conn, _ := Connect(n, ch)
 278         //
 279         ms := "A message"
 280         d := tdest("/queue/subunsub.basiconn.r10.01")
 281         sh := Headers{HK_DESTINATION, d}
 282         _ = conn.Send(sh, ms)
 283         //
 284         sbh := sh
 285         sc, e := conn.Subscribe(sbh)
 286         if e != nil {
 287                 t.Fatalf("Expected no subscribe error, got [%v]\n", e)
 288         }
 289         if sc == nil {
 290                 t.Fatalf("Expected subscribe channel, got [nil]\n")
 291         }
 292
 293         // Read MessageData
 294         var md MessageData
 295         select {
 296         case md = <-sc:
 297         case md = <-conn.MessageData:
 298                 t.Fatalf("read channel error:  expected [nil], got: [%v]\n",
 299                         md.Message.Command)
 300         }
 301
 302         //
 303         if md.Error != nil {
 304                 t.Fatalf("Expected no message data error, got [%v]\n", md.Error)
 305         }
 306         mdm := md.Message
 307         rd := mdm.Headers.Value(HK_DESTINATION)
 308         if rd != d {
 309                 t.Fatalf("Expected destination [%v], got [%v]\n", d, rd)
 310         }
 311         //
 312         e = conn.Unsubscribe(sbh)
 313         if e != nil {
 314                 t.Fatalf("Expected no unsubscribe error, got [%v]\n", e)
 315         }
 316         //
 317         _ = conn.Disconnect(empty_headers)
 318         _ = closeConn(t, n)
 319 }
 320
 321 /*
 322         Test establishSubscription.
 323 */
 324 func TestSubEstablishSubscription(t *testing.T) {
 325
 326         n, _ := openConn(t)
 327         ch := check11(TEST_HEADERS)
 328         conn, _ := Connect(n, ch)
 329         //
 330         d := tdest("/queue/estabsub.01")
 331         sbh := Headers{HK_DESTINATION, d}
 332         // First time
 333         s, e := conn.Subscribe(sbh)
 334         if e != nil {
 335                 t.Fatalf("Expected no subscribe error, got [%v]\n", e)
 336         }
 337         if s == nil {
 338                 t.Fatalf("Expected subscribe channel, got [nil]\n")
 339         }
 340         //
 341         _ = conn.Disconnect(empty_headers)
 342         _ = closeConn(t, n)
 343 }
 344
 345 /*
 346         Test unsubscribe, set subscribe channel capacity.
 347 */
 348 func TestSubSetCap(t *testing.T) {
 349         if os.Getenv("STOMP_TEST11p") == "" {
 350                 t.Skip("TestSubSetCap norun, need 1.1+")
 351         }
 352
 353         //
 354         n, _ := openConn(t)
 355         ch := check11(TEST_HEADERS)
 356         conn, _ := Connect(n, ch)
 357         //
 358         p := 25
 359         conn.SetSubChanCap(p)
 360         r := conn.SubChanCap()
 361         if r != p {
 362                 t.Fatalf("Expected get capacity [%v], got [%v]\n", p, r)
 363         }
 364         //
 365         d := tdest("/queue/subsetcap.basiconn.01")
 366         h := Headers{HK_DESTINATION, d, HK_ID, d}
 367         s, e := conn.Subscribe(h)
 368         if e != nil {
 369                 t.Fatalf("Expected no subscribe error, got [%v]\n", e)
 370         }
 371         if s == nil {
 372                 t.Fatalf("Expected subscribe channel, got [nil]\n")
 373         }
 374         if cap(s) != p {
 375                 t.Fatalf("Expected subchan capacity [%v], got [%v]\n", p, cap(s))
 376         }
 377         //
 378         e = conn.Unsubscribe(h)
 379         if e != nil {
 380                 t.Fatalf("Expected no unsubscribe error, got [%v]\n", e)
 381         }
 382         //
 383         _ = conn.Disconnect(empty_headers)
 384         _ = closeConn(t, n)
 385 }