Do live pulsar produce/consume in tests when configured

.travis.yml

@@ -1,5 +1,7 @@
 dist: bionic # ubuntu 18.04
 sudo: false
+services:
+  - docker
 language: ruby
 rvm:
   - 2.4.9 # oldest pre-install on bionic
@@ -12,5 +14,14 @@ install:
   - sudo apt-get -y install ./apache-pulsar-client-dev.deb
   - gem install bundler -v 1.16.1
   - bundle install
+before_script:
+  - docker run --name pulsar -d -p 6650:6650 -p 8080:8080 apachepulsar/pulsar:latest bin/pulsar standalone
+  - docker exec pulsar bash -c 'ready (){ (exec > /dev/tcp/$1/$2); }; i=0; while ! ready localhost 8080; do sleep 1; i=$((i+1)); [[ $i -gt 10 ]] && break; done'
+  - docker exec pulsar bin/pulsar-admin tenants create ruby-client
+  - docker exec pulsar bin/pulsar-admin namespaces create ruby-client/tests
+env:
+  global:
+    PULSAR_BROKER_URI: pulsar://localhost:6650
+    PULSAR_CLIENT_RUBY_TEST_NAMESPACE: ruby-client/tests
 script:
   - rake

spec/pulsar/client_spec.rb

@@ -21,4 +21,31 @@
   it "has a version number" do
     expect(Pulsar::Client::VERSION).not_to be nil
   end
+
+  context 'with pulsar' do
+    let(:broker_uri) { ENV['PULSAR_BROKER_URI'] }
+    let(:namespace) { ENV['PULSAR_CLIENT_RUBY_TEST_NAMESPACE'].to_s.sub(%r{^[-a-z]:/+}, '') }
+    let(:configured?) { broker_uri && !namespace.empty? }
+    let(:client) { Pulsar::Client.from_environment(broker_uri: broker_uri) }
+    let(:topic) { "non-persistent://#{namespace}/test#{sprintf "%06d", rand(1_000_000)}" }
+    let(:producer) { client.create_producer(topic) }
+    let(:subscription_name) { "#{topic}-consumer" }
+    let(:timeout_ms) { 10_000 }
+
+    before(:each) do
+      skip('Live Pulsar tests not configured: Set PULSAR_CLIENT_RUBY_TEST_NAMESPACE to enable') unless configured?
+    end
+
+    after(:each) do
+      # Close any producers/consumers to avoid test pollution.
+      client.close
+    end
+
+    it "can consume a single topic" do
+      consumer = client.subscribe(topic, subscription_name)
+      t = Thread.new { consumer.receive(timeout_ms).data }
+      client.create_producer(topic).send("single")
+      expect(t.join.value).to eq("single")
+    end
+  end
 end