diff --git a/lib/tapfall/errors.rb b/lib/tapfall/errors.rb
index b508bb2..4ec8b3f 100644
--- a/lib/tapfall/errors.rb
+++ b/lib/tapfall/errors.rb
@@ -1,4 +1,7 @@
 module Tapfall
+  class ConfigError < StandardError
+  end
+
   class DecodeError < StandardError
   end
 end
diff --git a/lib/tapfall/stream.rb b/lib/tapfall/stream.rb
index 14aff8e..f6a26c8 100644
--- a/lib/tapfall/stream.rb
+++ b/lib/tapfall/stream.rb
@@ -3,22 +3,38 @@ require_relative 'version'
 require_relative 'messages/tap_message'
 
 class Tapfall::Stream < Skyfall::Stream
-  def initialize(server)
+  def initialize(server, options = {})
     super(server)
 
+    @options = options
     @root_url = ensure_empty_path(@root_url)
+    @ack = true unless options[:ack] == false
   end
 
-  def handle_message(msg)
-    data = msg.data
+  def connect
+    if @ack && @handlers[:message].nil?
+      raise ConfigError, "The on(:message) handler must be set unless :ack => false option is passed"
+    end
+
+    super
+  end
+
+  def handle_message(packet)
+    data = packet.data
     @handlers[:raw_message]&.call(data)
 
     if @handlers[:message]
       tap_message = Tapfall::TapMessage.new(data)
       @handlers[:message].call(tap_message)
+      send_ack(tap_message) if @ack
     end
   end
 
+  def send_ack(msg)
+    json = JSON.generate(type: 'ack', id: msg.id)
+    send_data(json)
+  end
+
   private
 
   def build_websocket_url