Add protocol to reader and writer state, and pass it around in enough places to let us switch based on protocol. Remove a couple of unused exports while we're there.

5 files changed, 120 insertions, 93 deletions

diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index 06297c69f3..cff17e080a 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -36,7 +36,8 @@
 
 -record(vhost, {virtual_host, dummy}).
 
--record(connection, {user, timeout_sec, frame_max, vhost, client_properties}).
+-record(connection, {user, timeout_sec, frame_max, vhost, client_properties,
+                     protocol}).
 
 -record(content,
         {class_id,
@@ -173,6 +174,7 @@
                   explanation :: string(),
                   method      :: atom()}).
 
+-type(protocol() :: atom()).
 -endif.
 
 %%----------------------------------------------------------------------------
diff --git a/src/rabbit_binary_generator.erl b/src/rabbit_binary_generator.erl
index cd8b44d15f..e29f9fcbc7 100644
--- a/src/rabbit_binary_generator.erl
+++ b/src/rabbit_binary_generator.erl
@@ -41,7 +41,7 @@
 % See definition of check_empty_content_body_frame_size/0, an assertion called at startup.
 -define(EMPTY_CONTENT_BODY_FRAME_SIZE, 8).
 
--export([build_simple_method_frame/2,
+-export([build_simple_method_frame/3,
          build_simple_content_frames/3,
          build_heartbeat_frame/0]).
 -export([generate_table/1, encode_properties/2]).
@@ -56,8 +56,8 @@
 
 -type(frame() :: [binary()]).
 
--spec(build_simple_method_frame/2 ::
-      (channel_number(), amqp_method_record()) -> frame()).
+-spec(build_simple_method_frame/3 ::
+      (channel_number(), amqp_method_record(), protocol()) -> frame()).
 -spec(build_simple_content_frames/3 ::
       (channel_number(), content(), non_neg_integer()) -> [frame()]).
 -spec(build_heartbeat_frame/0 :: () -> frame()).
@@ -71,17 +71,18 @@
 
 %%----------------------------------------------------------------------------
 
-build_simple_method_frame(ChannelInt, MethodRecord) ->
+build_simple_method_frame(ChannelInt, MethodRecord, Protocol) ->
     MethodFields = rabbit_framing:encode_method_fields(MethodRecord),
-    MethodName = adjust_close(rabbit_misc:method_record_type(MethodRecord)),
+    MethodName = adjust_close(rabbit_misc:method_record_type(MethodRecord),
+                              Protocol),
     {ClassId, MethodId} = rabbit_framing:method_id(MethodName),
     create_frame(1, ChannelInt, [<<ClassId:16, MethodId:16>>, MethodFields]).
 
-adjust_close('connection.close') ->
+adjust_close('connection.close', protocol_08) ->
     'connection.close08';
-adjust_close('connection.close_ok') ->
+adjust_close('connection.close_ok', protocol_08) ->
     'connection.close08_ok';
-adjust_close(MethodName) ->
+adjust_close(MethodName, _Protocol) ->
     MethodName.
 
 build_simple_content_frames(ChannelInt,
diff --git a/src/rabbit_framing_channel.erl b/src/rabbit_framing_channel.erl
index 490435835d..90856cdf3f 100644
--- a/src/rabbit_framing_channel.erl
+++ b/src/rabbit_framing_channel.erl
@@ -32,20 +32,20 @@
 -module(rabbit_framing_channel).
 -include("rabbit.hrl").
 
--export([start_link/2, process/2, shutdown/1]).
+-export([start_link/3, process/2, shutdown/1]).
 
 %% internal
--export([mainloop/1]).
+-export([mainloop/2]).
 
 %%--------------------------------------------------------------------
 
-start_link(StartFun, StartArgs) ->
+start_link(StartFun, StartArgs, Protocol) ->
     spawn_link(
       fun () ->
               %% we trap exits so that a normal termination of the
               %% channel or reader process terminates us too.
               process_flag(trap_exit, true),
-              mainloop(apply(StartFun, StartArgs))
+              mainloop(apply(StartFun, StartArgs), Protocol)
       end).
 
 process(Pid, Frame) ->
@@ -72,30 +72,30 @@ read_frame(ChannelPid) ->
         Msg                    -> exit({unexpected_message, Msg})
     end.
 
-mainloop(ChannelPid) ->
+mainloop(ChannelPid, Protocol) ->
     {method, MethodName, FieldsBin} = read_frame(ChannelPid),
-    Method = decode_method_fields(MethodName, FieldsBin),
+    Method = decode_method_fields(MethodName, FieldsBin, Protocol),
     case rabbit_framing:method_has_content(MethodName) of
         true  -> rabbit_channel:do(ChannelPid, Method,
                                    collect_content(ChannelPid, MethodName));
         false -> rabbit_channel:do(ChannelPid, Method)
     end,
-    ?MODULE:mainloop(ChannelPid).
+    ?MODULE:mainloop(ChannelPid, Protocol).
 
 %% Handle 0-8 version of channel.tune-ok. In 0-9-1 it gained a longstr
 %% "deprecated_channel_id".
-decode_method_fields('channel.tune_ok', FieldsBin) ->
+decode_method_fields('channel.tune_ok', FieldsBin, protocol_08) ->
     Len = 0,
     rabbit_framing:decode_method_fields(
       'channel.tune_ok', <<FieldsBin/binary, Len:32/unsigned>>);
 %% Handle 0-8 version of basic.consume. In 0-9-1 it gained a table
 %% "filter".
-decode_method_fields('basic.consume', FieldsBin) ->
+decode_method_fields('basic.consume', FieldsBin, protocol_08) ->
     T = rabbit_binary_generator:generate_table([]),
     TLen = size(T),
     rabbit_framing:decode_method_fields(
       'basic.consume', <<FieldsBin/binary, TLen:32/unsigned, T:TLen/binary>>);
-decode_method_fields(MethodName, FieldsBin) ->
+decode_method_fields(MethodName, FieldsBin, _Protocol) ->
     rabbit_framing:decode_method_fields(MethodName, FieldsBin).
 
 collect_content(ChannelPid, MethodName) ->
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index f83f932db9..e2d4880b94 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -41,8 +41,6 @@
 
 -export([server_properties/0]).
 
--export([analyze_frame/2]).
-
 -import(gen_tcp).
 -import(fprof).
 -import(inet).
@@ -244,7 +242,8 @@ start_connection(Parent, Deb, Sock, SockTransform) ->
                                       timeout_sec = ?HANDSHAKE_TIMEOUT,
                                       frame_max = ?FRAME_MIN_SIZE,
                                       vhost = none,
-                                      client_properties = none},
+                                      client_properties = none,
+                                      protocol = unknown},
                                     callback = uninitialized_callback,
                                     recv_ref = none,
                                     connection_state = pre_init,
@@ -432,7 +431,8 @@ wait_for_channel_termination(N, TimerRef) ->
     end.
 
 maybe_close(State = #v1{connection_state = closing,
-                        queue_collector = Collector}) ->
+                        queue_collector = Collector,
+                        connection = #connection{protocol = Protocol}}) ->
     case all_channels() of
         [] ->
             %% Spec says "Exclusive queues may only be accessed by the current
@@ -440,16 +440,19 @@ maybe_close(State = #v1{connection_state = closing,
             %% This does not strictly imply synchrony, but in practice it seems
             %% to be what people assume.
             rabbit_reader_queue_collector:delete_all(Collector),
-            ok = send_on_channel0(State#v1.sock, #'connection.close_ok'{}),
+            ok = send_on_channel0(State#v1.sock, #'connection.close_ok'{},
+                                  Protocol),
             close_connection(State);
         _  -> State
     end;
 maybe_close(State) ->
     State.
 
-handle_frame(Type, 0, Payload, State = #v1{connection_state = CS})
+handle_frame(Type, 0, Payload,
+             State = #v1{connection_state = CS,
+                         connection = #connection{protocol = Protocol}})
   when CS =:= closing; CS =:= closed ->
-    case analyze_frame(Type, Payload) of
+    case analyze_frame(Type, Payload, Protocol) of
         {method, MethodName, FieldsBin} ->
             handle_method0(MethodName, FieldsBin, State);
         _Other -> State
@@ -457,16 +460,18 @@ handle_frame(Type, 0, Payload, State = #v1{connection_state = CS})
 handle_frame(_Type, _Channel, _Payload, State = #v1{connection_state = CS})
   when CS =:= closing; CS =:= closed ->
     State;
-handle_frame(Type, 0, Payload, State) ->
-    case analyze_frame(Type, Payload) of
+handle_frame(Type, 0, Payload,
+             State = #v1{connection = #connection{protocol = Protocol}}) ->
+    case analyze_frame(Type, Payload, Protocol) of
         error     -> throw({unknown_frame, 0, Type, Payload});
         heartbeat -> State;
         {method, MethodName, FieldsBin} ->
             handle_method0(MethodName, FieldsBin, State);
         Other -> throw({unexpected_frame_on_channel0, Other})
     end;
-handle_frame(Type, Channel, Payload, State) ->
-    case analyze_frame(Type, Payload) of
+handle_frame(Type, Channel, Payload,
+             State = #v1{connection = #connection{protocol = Protocol}}) ->
+    case analyze_frame(Type, Payload, Protocol) of
         error         -> throw({unknown_frame, Channel, Type, Payload});
         heartbeat     -> throw({unexpected_heartbeat_frame, Channel});
         AnalyzedFrame ->
@@ -501,24 +506,28 @@ handle_frame(Type, Channel, Payload, State) ->
             end
     end.
 
-analyze_frame(?FRAME_METHOD, <<ClassId:16, MethodId:16, MethodFields/binary>>) ->
-    {method, adjust_close(rabbit_framing:lookup_method_name({ClassId,
-                                                             MethodId})),
+analyze_frame(?FRAME_METHOD, <<ClassId:16, MethodId:16, MethodFields/binary>>,
+              Protocol) ->
+    {method, adjust_close(
+               rabbit_framing:lookup_method_name({ClassId, MethodId}),
+               Protocol),
      MethodFields};
-analyze_frame(?FRAME_HEADER, <<ClassId:16, Weight:16, BodySize:64, Properties/binary>>) ->
+analyze_frame(?FRAME_HEADER,
+              <<ClassId:16, Weight:16, BodySize:64, Properties/binary>>,
+              _Protocol) ->
     {content_header, ClassId, Weight, BodySize, Properties};
-analyze_frame(?FRAME_BODY, Body) ->
+analyze_frame(?FRAME_BODY, Body, _Protocol) ->
     {content_body, Body};
-analyze_frame(?FRAME_HEARTBEAT, <<>>) ->
+analyze_frame(?FRAME_HEARTBEAT, <<>>, _Protocol) ->
     heartbeat;
-analyze_frame(_Type, _Body) ->
+analyze_frame(_Type, _Body, _Protocol) ->
     error.
 
-adjust_close('connection.close08') ->
+adjust_close('connection.close08', protocol_08) ->
     'connection.close';
-adjust_close('connection.close08_ok') ->
+adjust_close('connection.close08_ok', protocol_08) ->
     'connection.close_ok';
-adjust_close(MethodName) ->
+adjust_close(MethodName, _Protocol) ->
     MethodName.
 
 handle_input(frame_header, <<Type:8,Channel:16,PayloadSize:32>>, State) ->
@@ -540,6 +549,7 @@ handle_input(handshake, <<"AMQP",1,1,ProtocolMajor,ProtocolMinor>>,
     case check_version({ProtocolMajor, ProtocolMinor},
                        {?PROTOCOL_VERSION_MAJOR, ?PROTOCOL_VERSION_MINOR}) of
         true ->
+            Protocol = protocol_08,
             ok = send_on_channel0(
                    Sock,
                    #'connection.start'{
@@ -547,9 +557,11 @@ handle_input(handshake, <<"AMQP",1,1,ProtocolMajor,ProtocolMinor>>,
                      version_minor = ?PROTOCOL_VERSION_MINOR,
                      server_properties = server_properties(),
                      mechanisms = <<"PLAIN AMQPLAIN">>,
-                     locales = <<"en_US">> }),
+                     locales = <<"en_US">> },
+                  Protocol),
             {State#v1{connection = Connection#connection{
-                                     timeout_sec = ?NORMAL_TIMEOUT},
+                                     timeout_sec = ?NORMAL_TIMEOUT,
+                                     protocol = Protocol},
                       connection_state = starting},
              frame_header, 7};
         false ->
@@ -604,7 +616,8 @@ handle_method0(#'connection.start_ok'{mechanism = Mechanism,
                                       response = Response,
                                       client_properties = ClientProperties},
                State = #v1{connection_state = starting,
-                           connection = Connection,
+                           connection = Connection =
+                               #connection{protocol = Protocol},
                            sock = Sock}) ->
     User = rabbit_access_control:check_login(Mechanism, Response),
     ok = send_on_channel0(
@@ -612,7 +625,8 @@ handle_method0(#'connection.start_ok'{mechanism = Mechanism,
            #'connection.tune'{channel_max = 0,
                               %% set to zero once QPid fix their negotiation
                               frame_max = 131072,
-                              heartbeat = 0}),
+                              heartbeat = 0},
+           Protocol),
     State#v1{connection_state = tuning,
              connection = Connection#connection{
                             user = User,
@@ -634,13 +648,15 @@ handle_method0(#'connection.tune_ok'{channel_max = _ChannelMax,
 handle_method0(#'connection.open'{virtual_host = VHostPath},
                State = #v1{connection_state = opening,
                            connection = Connection = #connection{
-                                          user = User},
+                                          user = User,
+                                          protocol = Protocol},
                            sock = Sock}) ->
     ok = rabbit_access_control:check_vhost_access(User, VHostPath),
     NewConnection = Connection#connection{vhost = VHostPath},
     ok = send_on_channel0(
            Sock,
-           #'connection.open_ok'{deprecated_known_hosts = <<>>}),
+           #'connection.open_ok'{deprecated_known_hosts = <<>>},
+           Protocol),
     State#v1{connection_state = running,
              connection = NewConnection};
 handle_method0(#'connection.close'{},
@@ -658,8 +674,8 @@ handle_method0(_Method, #v1{connection_state = S}) ->
     rabbit_misc:protocol_error(
       channel_error, "unexpected method in connection state ~w", [S]).
 
-send_on_channel0(Sock, Method) ->
-    ok = rabbit_writer:internal_send_command(Sock, 0, Method).
+send_on_channel0(Sock, Method, Protocol) ->
+    ok = rabbit_writer:internal_send_command(Sock, 0, Method, Protocol).
 
 %%--------------------------------------------------------------------------
 
@@ -711,16 +727,19 @@ i(Item, #v1{}) ->
 
 %%--------------------------------------------------------------------------
 
-send_to_new_channel(Channel, AnalyzedFrame,
-                    State = #v1{queue_collector = Collector}) ->
+send_to_new_channel(Channel, AnalyzedFrame, State =
+                        #v1{queue_collector = Collector,
+                            connection = #connection{protocol = Protocol}}) ->
     #v1{sock = Sock, connection = #connection{
                        frame_max = FrameMax,
                        user = #user{username = Username},
-                       vhost = VHost}} = State,
-    WriterPid = rabbit_writer:start(Sock, Channel, FrameMax),
+                       vhost = VHost,
+                       protocol = Protocol}} = State,
+    WriterPid = rabbit_writer:start(Sock, Channel, FrameMax, Protocol),
     ChPid = rabbit_framing_channel:start_link(
               fun rabbit_channel:start_link/6,
-              [Channel, self(), WriterPid, Username, VHost, Collector]),
+              [Channel, self(), WriterPid, Username, VHost, Collector],
+              Protocol),
     put({channel, Channel}, {chpid, ChPid}),
     put({chpid, ChPid}, {channel, Channel}),
     ok = rabbit_framing_channel:process(ChPid, AnalyzedFrame).
@@ -736,7 +755,8 @@ handle_exception(State = #v1{connection_state = CS}, Channel, Reason) ->
     log_channel_error(CS, Channel, Reason),
     send_exception(State, Channel, Reason).
 
-send_exception(State, Channel, Reason) ->
+send_exception(State = #v1{connection = #connection{protocol = Protocol}},
+               Channel, Reason) ->
     {ShouldClose, CloseChannel, CloseMethod} = map_exception(Channel, Reason),
     NewState = case ShouldClose of
                    true  -> terminate_channels(),
@@ -744,7 +764,7 @@ send_exception(State, Channel, Reason) ->
                    false -> close_channel(Channel, State)
                end,
     ok = rabbit_writer:internal_send_command(
-           NewState#v1.sock, CloseChannel, CloseMethod),
+           NewState#v1.sock, CloseChannel, CloseMethod, Protocol),
     NewState.
 
 map_exception(Channel, Reason) ->
diff --git a/src/rabbit_writer.erl b/src/rabbit_writer.erl
index 3d10dc121e..375e540e7f 100644
--- a/src/rabbit_writer.erl
+++ b/src/rabbit_writer.erl
@@ -33,14 +33,14 @@
 -include("rabbit.hrl").
 -include("rabbit_framing.hrl").
 
--export([start/3, start_link/3, shutdown/1, mainloop/1]).
+-export([start/4, shutdown/1, mainloop/1]).
 -export([send_command/2, send_command/3, send_command_and_signal_back/3,
          send_command_and_signal_back/4, send_command_and_notify/5]).
--export([internal_send_command/3, internal_send_command/5]).
+-export([internal_send_command/4, internal_send_command/6]).
 
 -import(gen_tcp).
 
--record(wstate, {sock, channel, frame_max}).
+-record(wstate, {sock, channel, frame_max, protocol}).
 
 -define(HIBERNATE_AFTER, 5000).
 
@@ -48,8 +48,8 @@
 
 -ifdef(use_specs).
 
--spec(start/3 :: (socket(), channel_number(), non_neg_integer()) -> pid()).
--spec(start_link/3 :: (socket(), channel_number(), non_neg_integer()) -> pid()).
+-spec(start/4 ::
+        (socket(), channel_number(), non_neg_integer(), protocol()) -> pid()).
 -spec(send_command/2 :: (pid(), amqp_method_record()) -> 'ok').
 -spec(send_command/3 :: (pid(), amqp_method_record(), content()) -> 'ok').
 -spec(send_command_and_signal_back/3 :: (pid(), amqp_method(), pid()) -> 'ok').
@@ -57,25 +57,21 @@
       (pid(), amqp_method(), content(), pid()) -> 'ok').
 -spec(send_command_and_notify/5 ::
       (pid(), pid(), pid(), amqp_method_record(), content()) -> 'ok').
--spec(internal_send_command/3 ::
-      (socket(), channel_number(), amqp_method_record()) -> 'ok').
--spec(internal_send_command/5 ::
+-spec(internal_send_command/4 ::
+      (socket(), channel_number(), amqp_method_record(), protocol()) -> 'ok').
+-spec(internal_send_command/6 ::
       (socket(), channel_number(), amqp_method_record(),
-       content(), non_neg_integer()) -> 'ok').
+       content(), non_neg_integer(), protocol()) -> 'ok').
 
 -endif.
 
 %%----------------------------------------------------------------------------
 
-start(Sock, Channel, FrameMax) ->
+start(Sock, Channel, FrameMax, Protocol) ->
     spawn(?MODULE, mainloop, [#wstate{sock = Sock,
                                       channel = Channel,
-                                      frame_max = FrameMax}]).
-
-start_link(Sock, Channel, FrameMax) ->
-    spawn_link(?MODULE, mainloop, [#wstate{sock = Sock,
-                                           channel = Channel,
-                                           frame_max = FrameMax}]).
+                                      frame_max = FrameMax,
+                                      protocol = Protocol}]).
 
 mainloop(State) ->
     receive
@@ -85,35 +81,40 @@ mainloop(State) ->
     end.
 
 handle_message({send_command, MethodRecord},
-               State = #wstate{sock = Sock, channel = Channel}) ->
-    ok = internal_send_command_async(Sock, Channel, MethodRecord),
+               State = #wstate{sock = Sock, channel = Channel,
+                               protocol = Protocol}) ->
+    ok = internal_send_command_async(Sock, Channel, MethodRecord, Protocol),
     State;
 handle_message({send_command, MethodRecord, Content},
                State = #wstate{sock = Sock,
                                channel = Channel,
-                               frame_max = FrameMax}) ->
+                               frame_max = FrameMax,
+                               protocol = Protocol}) ->
     ok = internal_send_command_async(Sock, Channel, MethodRecord,
-                                     Content, FrameMax),
+                                     Content, FrameMax, Protocol),
     State;
 handle_message({send_command_and_signal_back, MethodRecord, Parent},
-               State = #wstate{sock = Sock, channel = Channel}) ->
-    ok = internal_send_command_async(Sock, Channel, MethodRecord),
+               State = #wstate{sock = Sock, channel = Channel,
+                               protocol = Protocol}) ->
+    ok = internal_send_command_async(Sock, Channel, MethodRecord, Protocol),
     Parent ! rabbit_writer_send_command_signal,
     State;
 handle_message({send_command_and_signal_back, MethodRecord, Content, Parent},
                State = #wstate{sock = Sock,
                                channel = Channel,
-                               frame_max = FrameMax}) ->
+                               frame_max = FrameMax,
+                               protocol = Protocol}) ->
     ok = internal_send_command_async(Sock, Channel, MethodRecord,
-                                     Content, FrameMax),
+                                     Content, FrameMax, Protocol),
     Parent ! rabbit_writer_send_command_signal,
     State;
 handle_message({send_command_and_notify, QPid, ChPid, MethodRecord, Content},
                State = #wstate{sock = Sock,
                                channel = Channel,
-                               frame_max = FrameMax}) ->
+                               frame_max = FrameMax,
+                               protocol = Protocol}) ->
     ok = internal_send_command_async(Sock, Channel, MethodRecord,
-                                     Content, FrameMax),
+                                     Content, FrameMax, Protocol),
     rabbit_amqqueue:notify_sent(QPid, ChPid),
     State;
 handle_message({inet_reply, _, ok}, State) ->
@@ -153,16 +154,17 @@ shutdown(W) ->
 
 %---------------------------------------------------------------------------
 
-assemble_frames(Channel, MethodRecord) ->
+assemble_frames(Channel, MethodRecord, Protocol) ->
     ?LOGMESSAGE(out, Channel, MethodRecord, none),
-    rabbit_binary_generator:build_simple_method_frame(Channel, MethodRecord).
+    rabbit_binary_generator:build_simple_method_frame(Channel, MethodRecord,
+                                                      Protocol).
 
-assemble_frames(Channel, MethodRecord, Content, FrameMax) ->
+assemble_frames(Channel, MethodRecord, Content, FrameMax, Protocol) ->
     ?LOGMESSAGE(out, Channel, MethodRecord, Content),
     MethodName = rabbit_misc:method_record_type(MethodRecord),
     true = rabbit_framing:method_has_content(MethodName), % assertion
     MethodFrame = rabbit_binary_generator:build_simple_method_frame(
-                    Channel, MethodRecord),
+                    Channel, MethodRecord, Protocol),
     ContentFrames = rabbit_binary_generator:build_simple_content_frames(
                       Channel, Content, FrameMax),
     [MethodFrame | ContentFrames].
@@ -171,12 +173,13 @@ tcp_send(Sock, Data) ->
     rabbit_misc:throw_on_error(inet_error,
                                fun () -> rabbit_net:send(Sock, Data) end).
 
-internal_send_command(Sock, Channel, MethodRecord) ->
-    ok = tcp_send(Sock, assemble_frames(Channel, MethodRecord)).
+internal_send_command(Sock, Channel, MethodRecord, Protocol) ->
+    ok = tcp_send(Sock, assemble_frames(Channel, MethodRecord, Protocol)).
 
-internal_send_command(Sock, Channel, MethodRecord, Content, FrameMax) ->
+internal_send_command(Sock, Channel, MethodRecord, Content, FrameMax,
+                      Protocol) ->
     ok = tcp_send(Sock, assemble_frames(Channel, MethodRecord,
-                                        Content, FrameMax)).
+                                        Content, FrameMax, Protocol)).
 
 %% gen_tcp:send/2 does a selective receive of {inet_reply, Sock,
 %% Status} to obtain the result. That is bad when it is called from
@@ -196,13 +199,14 @@ internal_send_command(Sock, Channel, MethodRecord, Content, FrameMax) ->
 %% Also note that the port has bounded buffers and port_command blocks
 %% when these are full. So the fact that we process the result
 %% asynchronously does not impact flow control.
-internal_send_command_async(Sock, Channel, MethodRecord) ->
-    true = port_cmd(Sock, assemble_frames(Channel, MethodRecord)),
+internal_send_command_async(Sock, Channel, MethodRecord, Protocol) ->
+    true = port_cmd(Sock, assemble_frames(Channel, MethodRecord, Protocol)),
     ok.
 
-internal_send_command_async(Sock, Channel, MethodRecord, Content, FrameMax) ->
+internal_send_command_async(Sock, Channel, MethodRecord, Content, FrameMax,
+                            Protocol) ->
     true = port_cmd(Sock, assemble_frames(Channel, MethodRecord,
-                                              Content, FrameMax)),
+                                              Content, FrameMax, Protocol)),
     ok.
 
 port_cmd(Sock, Data) ->