ack_test.go

   1 //
   2 // Copyright © 2011-2019 Guy M. Allard
   3 //
   4 // Licensed under the Apache License, Veridon 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 permisidons and
  14 // limitations under the License.
  15 //
  16
  17 package stompngo
  18
  19 import (
  20         "fmt"
  21         "log"
  22         "os"
  23         "testing"
  24         "time"
  25 )
  26
  27 var _ = fmt.Println
  28
  29 /*
  30         Test Ack errors.
  31 */
  32 func TestAckErrors(t *testing.T) {
  33         n, _ = openConn(t)
  34         ch := login_headers
  35         conn, e = Connect(n, ch)
  36         if e != nil {
  37                 t.Fatalf("TestAckErrors CONNECT expected nil, got %v\n", e)
  38         }
  39         //
  40         for _, tv := range terrList {
  41                 conn.protocol = tv.proto // Fake it
  42                 e = conn.Ack(tv.headers)
  43                 if e != tv.errval {
  44                         t.Fatalf("ACK -%s- expected error [%v], got [%v]\n",
  45                                 tv.proto, tv.errval, e)
  46                 }
  47         }
  48         checkReceived(t, conn, false)
  49         e = conn.Disconnect(empty_headers)
  50         checkDisconnectError(t, e)
  51         _ = closeConn(t, n)
  52 }
  53
  54 /*
  55         Test Ack Same Connection.
  56 */
  57 func TestAckSameConn(t *testing.T) {
  58         for _, sp := range Protocols() {
  59                 n, _ = openConn(t)
  60                 ch := login_headers
  61                 ch = headersProtocol(ch, sp)
  62                 conn, e = Connect(n, ch)
  63                 if e != nil {
  64                         t.Fatalf("TestAckSameConn CONNECT expected nil, got %v\n", e)
  65                 }
  66                 //
  67                 // Basic headers
  68                 wh := Headers{HK_DESTINATION,
  69                         tdest(TEST_TDESTPREF + "acksc1-" + conn.Protocol())}
  70                 // Subscribe Headers
  71                 sbh := wh.Add(HK_ACK, AckModeClient)
  72                 id := TEST_TDESTPREF + "acksc1.chkprotocol-" + conn.Protocol()
  73                 sbh = sbh.Add(HK_ID, id) // Always use an 'id'
  74                 ms := "acksc1 message 1"
  75                 //
  76                 // Subscribe
  77                 sc, e = conn.Subscribe(sbh)
  78                 if e != nil {
  79                         t.Fatalf("TestAckSameConn SUBSCRIBE expected [nil], got: [%v]\n", e)
  80                 }
  81
  82                 //
  83                 // Send
  84                 sh := wh.Clone()
  85                 // For RabbitMQ and STOMP 1.0, do not add current-time header, where the
  86                 // value contains ':' characters.
  87                 switch conn.Protocol() {
  88                 case SPL_10:
  89                         if os.Getenv("STOMP_RMQ") == "" {
  90                                 sh = sh.Add("current-time", time.Now().String()) // The added header value has ':' characters
  91                         }
  92                 default:
  93                         sh = sh.Add("current-time", time.Now().String()) // The added header value has ':' characters
  94                 }
  95                 e = conn.Send(sh, ms)
  96                 if e != nil {
  97                         t.Fatalf("TestAckSameConn SEND expected [nil], got: [%v]\n", e)
  98                 }
  99                 //
 100                 // Read MessageData
 101                 select {
 102                 case md = <-sc:
 103                 case md = <-conn.MessageData:
 104                         t.Fatalf("TestAckSameConn read channel error:  expected [nil], got: [%v] msg: [%v] err: [%v]\n",
 105                                 md.Message.Command, md.Message, md.Error)
 106                 }
 107                 if md.Error != nil {
 108                         t.Fatalf("TestAckSameConn read error:  expected [nil], got: [%v]\n",
 109                                 md.Error)
 110                 }
 111                 if ms != md.Message.BodyString() {
 112                         t.Fatalf("TestAckSameConn message error: expected: [%v], got: [%v] Message: [%q]\n",
 113                                 ms, md.Message.BodyString(), md.Message)
 114                 }
 115                 // Ack headers
 116                 ah := Headers{}
 117                 if conn.Protocol() == SPL_12 {
 118                         ah = ah.Add(HK_ID, md.Message.Headers.Value(HK_ACK))
 119                 } else {
 120                         ah = ah.Add(HK_MESSAGE_ID, md.Message.Headers.Value(HK_MESSAGE_ID))
 121                 }
 122                 //
 123                 if conn.Protocol() == SPL_11 {
 124                         ah = ah.Add(HK_SUBSCRIPTION, id) // Always use subscription for 1.1
 125                 }
 126                 // Ack
 127                 e = conn.Ack(ah)
 128                 if e != nil {
 129                         t.Fatalf("ACK expected [nil], got: [%v]\n", e)
 130                 }
 131                 // Make sure Apollo Jira issue APLO-88 stays fixed.
 132                 select {
 133                 case md = <-sc:
 134                         t.Fatalf("TestAckSameConn RECEIVE not expected, got: [%v]\n", md)
 135                 default:
 136                 }
 137
 138                 // Unsubscribe
 139                 uh := wh.Add(HK_ID, id)
 140                 e = conn.Unsubscribe(uh)
 141                 if e != nil {
 142                         t.Fatalf("TestAckSameConn UNSUBSCRIBE expected [nil], got: [%v]\n", e)
 143                 }
 144
 145                 //
 146                 checkReceived(t, conn, false)
 147                 e = conn.Disconnect(empty_headers)
 148                 checkDisconnectError(t, e)
 149                 _ = closeConn(t, n)
 150         }
 151 }
 152
 153 /*
 154         Test Ack Different Connection.
 155 */
 156 func TestAckDiffConn(t *testing.T) {
 157
 158         for _, sp := range Protocols() {
 159                 n, _ = openConn(t)
 160                 ch := login_headers
 161                 ch = headersProtocol(ch, sp)
 162                 conn, e = Connect(n, ch)
 163                 if e != nil {
 164                         t.Fatalf("TestAckDiffConn CONNECT expected nil, got %v\n", e)
 165                 }
 166                 //
 167                 // Basic headers
 168                 wh := Headers{HK_DESTINATION,
 169                         tdest(TEST_TDESTPREF + "ackdc1-" + conn.Protocol())}
 170                 ms := "ackdc1 message 1"
 171                 // Send
 172                 sh := wh.Clone()
 173                 // For RabbitMQ and STOMP 1.0, do not add current-time header, where the
 174                 // value contains ':' characters.
 175                 switch conn.Protocol() {
 176                 case SPL_10:
 177                         if os.Getenv("STOMP_RMQ") == "" {
 178                                 sh = sh.Add("current-time", time.Now().String()) // The added header value has ':' characters
 179                         }
 180                 default:
 181                         sh = sh.Add("current-time", time.Now().String()) // The added header value has ':' characters
 182                 }
 183                 e = conn.Send(sh, ms)
 184                 if e != nil {
 185                         t.Fatalf("TestAckDiffConn SEND expected [nil], got: [%v]\n", e)
 186                 }
 187                 //
 188                 checkReceived(t, conn, false)
 189                 e = conn.Disconnect(empty_headers)
 190                 checkDisconnectError(t, e)
 191                 _ = closeConn(t, n)
 192                 //
 193                 n, _ = openConn(t)
 194                 ch = login_headers
 195                 ch = headersProtocol(ch, sp)
 196                 conn, e = Connect(n, ch) // Reconnect
 197                 if e != nil {
 198                         t.Fatalf("TestAckDiffConn Second Connect, expected no error, got:<%v>\n", e)
 199                 }
 200                 //
 201                 // Subscribe Headers
 202                 sbh := wh.Add(HK_ACK, AckModeClient)
 203                 id := "ackdc1.chkprotocol-" + conn.Protocol()
 204                 sbh = sbh.Add(HK_ID, id) // Always use an 'id'
 205                 // Subscribe
 206                 log.Printf("SUB Headers: [%q]\n", sbh)
 207                 sc, e = conn.Subscribe(sbh)
 208                 if e != nil {
 209                         t.Fatalf("TestAckDiffConn SUBSCRIBE expected [nil], got: [%v]\n", e)
 210                 }
 211                 // Read MessageData
 212                 select {
 213                 case md = <-sc:
 214                 case md = <-conn.MessageData:
 215                         t.Fatalf("TestAckDiffConn read channel error:  expected [nil], got: [%v], msg: [%v], err: [%v]\n",
 216                                 md.Message.Command, md.Message, md.Error)
 217                 }
 218                 if md.Error != nil {
 219                         t.Fatalf("read error:  expected [nil], got: [%v]\n", md.Error)
 220                 }
 221                 if ms != md.Message.BodyString() {
 222                         t.Fatalf("TestAckDiffConn message error: expected: [%v], got: [%v] Message: [%q]\n",
 223                                 ms, md.Message.BodyString(), md.Message)
 224                 }
 225                 // Ack headers
 226                 ah := Headers{}
 227                 if conn.Protocol() == SPL_12 {
 228                         ah = ah.Add(HK_ID, md.Message.Headers.Value(HK_ACK))
 229                 } else {
 230                         ah = ah.Add(HK_MESSAGE_ID, md.Message.Headers.Value(HK_MESSAGE_ID))
 231                 }
 232                 //
 233                 if conn.Protocol() == SPL_11 {
 234                         ah = ah.Add(HK_SUBSCRIPTION, id) // Always use subscription for 1.1
 235                 }
 236                 // Ack
 237                 e = conn.Ack(ah)
 238                 if e != nil {
 239                         t.Fatalf("TestAckDiffConn ACK expected [nil], got: [%v]\n", e)
 240                 }
 241                 // Make sure Apollo Jira issue APLO-88 stays fixed.
 242                 select {
 243                 case md = <-sc:
 244                         t.Fatalf("TestAckDiffConn RECEIVE not expected, got: [%v]\n", md)
 245                 default:
 246                 }
 247                 // Unsubscribe
 248                 uh := wh.Add(HK_ID, id)
 249                 e = conn.Unsubscribe(uh)
 250                 if e != nil {
 251                         t.Fatalf("TestAckDiffConn UNSUBSCRIBE expected [nil], got: [%v]\n", e)
 252                 }
 253                 //
 254                 checkReceived(t, conn, false)
 255                 e = conn.Disconnect(empty_headers)
 256                 checkDisconnectError(t, e)
 257                 _ = closeConn(t, n)
 258         }
 259 }