Merge pull request #46 from JuliaComputing/tan/checkstaleconn

check for stale connection while framing/chunking message to send

Author: tanmaykm

.github/workflows/ci.yml

@@ -0,0 +1,48 @@
+name: CI
+on:
+  push:
+    branches: [master]
+    tags: ["*"]
+  pull_request:
+jobs:
+  test:
+    name: Julia ${{ matrix.version }} - ${{ matrix.os }} - ${{ matrix.arch }} - ${{ github.event_name }}
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        version:
+          - '1.0'
+          - '1' # automatically expands to the latest stable 1.x release of Julia
+          - nightly
+        os:
+          - ubuntu-latest
+        arch:
+          - x64
+    services:
+      rabbitmq:
+        image: rabbitmq:3-management
+        ports:
+          - 5672:5672
+    steps:
+      - uses: actions/checkout@v2
+      - uses: julia-actions/setup-julia@v1
+        with:
+          version: ${{ matrix.version }}
+          arch: ${{ matrix.arch }}
+      - uses: actions/cache@v1
+        env:
+          cache-name: cache-artifacts
+        with:
+          path: ~/.julia/artifacts
+          key: ${{ runner.os }}-test-${{ env.cache-name }}-${{ hashFiles('**/Project.toml') }}
+          restore-keys: |
+            ${{ runner.os }}-test-${{ env.cache-name }}-
+            ${{ runner.os }}-test-
+            ${{ runner.os }}-
+      - uses: julia-actions/julia-buildpkg@v1
+      - uses: julia-actions/julia-runtest@v1
+      - uses: julia-actions/julia-processcoverage@v1
+      - uses: codecov/codecov-action@v1
+        with:
+          file: lcov.info

Project.toml

@@ -3,7 +3,7 @@ uuid = "79c8b4cd-a41a-55fa-907c-fab5288e1383"
 keywords = ["amqpclient", "rabbitmq", "amqp", "amqp-client", "message-queue"]
 license = "MIT"
 desc = "A Julia AMQP (Advanced Message Queuing Protocol) / RabbitMQ Client."
-version = "0.4.2"
+version = "0.4.3"
 
 [deps]
 Logging = "56ddb016-857b-54e1-b83d-db4d58db5568"

README.md

@@ -1,7 +1,7 @@
 # AMQPClient
 
-[![Build Status](https://travis-ci.org/JuliaComputing/AMQPClient.jl.svg?branch=master)](https://travis-ci.org/JuliaComputing/AMQPClient.jl)
-[![Coverage Status](https://coveralls.io/repos/JuliaComputing/AMQPClient.jl/badge.svg?branch=master&service=github)](https://coveralls.io/github/JuliaComputing/AMQPClient.jl?branch=master)
+[![Build Status](https://github.com/JuliaComputing/AMQPClient.jl/workflows/CI/badge.svg)](https://github.com/JuliaComputing/AMQPClient.jl/actions?query=workflow%3ACI+branch%3Amaster)
+[![codecov.io](http://codecov.io/github/JuliaComputing/AMQPClient.jl/coverage.svg?branch=master)](http://codecov.io/github/JuliaComputing/AMQPClient.jl?branch=master)
 
 A Julia [AMQP (Advanced Message Queuing Protocol)](http://www.amqp.org/) Client.
 

src/protocol.jl

@@ -267,13 +267,15 @@ get_property(c::Connection, s::Symbol, default) = get(c.properties, s, default)
 send(c::MessageChannel, f, msgframes::Vector=[]) = send(c.conn, f, msgframes)
 function send(c::Connection, f, msgframes::Vector=[])
     #uncomment to enable synchronization (not required till we have preemptive tasks or threads)
+    @debug("queing messageframes", nframes=length(msgframes))
     lck = take!(c.sendlck)
     try
         put!(c.sendq, TAMQPGenericFrame(f))
         for m in msgframes
             put!(c.sendq, TAMQPGenericFrame(m))
         end
     finally
+        @debug("queued messageframes", nqueued=length(c.sendq.data))
         put!(c.sendlck, lck)
     end
     nothing
@@ -292,17 +294,25 @@ function send(c::MessageChannel, payload::TAMQPMethodPayload, msg::Union{Message
         # send one or more message body frames
         offset = 1
         msglen = length(message.data)
+        framemax = c.conn.framemax
+        if framemax <= 0
+            errormsg = (c.conn.state == CONN_STATE_OPEN) ? "Unexpected framemax ($framemax) value for connection" : "Connection closed"
+            throw(AMQPClientException(errormsg))
+        end
+
         while offset <= msglen
-            msgend = min(msglen, offset + c.conn.framemax - 1)
+            msgend = min(msglen, offset + framemax - 1)
             bodypayload = TAMQPBodyPayload(message.data[offset:msgend])
             offset = msgend + 1
+            @debug("sending", msglen, offset)
             push!(msgframes, TAMQPContentBodyFrame(frameprop, bodypayload))
         end
 
         send(c, TAMQPMethodFrame(frameprop, payload), msgframes)
     else
         send(c, TAMQPMethodFrame(frameprop, payload))
     end
+    @debug("sent", methodname=method_name(payload))
 end
 
 # ----------------------------------------