THRIFT-2856 refactor erlang basic transports and unify interfaces
Client: Erlang
Patch: Alisdair Sullivan
This closes #288
diff --git a/lib/erl/src/thrift.app.src b/lib/erl/src/thrift.app.src
index 28b8cb5..1a23f0d 100644
--- a/lib/erl/src/thrift.app.src
+++ b/lib/erl/src/thrift.app.src
@@ -37,6 +37,7 @@
thrift_http_transport,
thrift_json_parser,
thrift_json_protocol,
+ thrift_membuffer_transport,
thrift_memory_buffer,
thrift_processor,
thrift_protocol,
diff --git a/lib/erl/src/thrift_buffered_transport.erl b/lib/erl/src/thrift_buffered_transport.erl
index d4d614e..e9d3fff 100644
--- a/lib/erl/src/thrift_buffered_transport.erl
+++ b/lib/erl/src/thrift_buffered_transport.erl
@@ -21,57 +21,78 @@
-behaviour(thrift_transport).
-%% API
--export([new/1, new_transport_factory/1]).
+%% constructor
+-export([new/1]).
+%% protocol callbacks
+-export([read/2, read_exact/2, write/2, flush/1, close/1]).
+%% legacy api
+-export([new_transport_factory/1]).
-%% thrift_transport callbacks
--export([write/2, read/2, flush/1, close/1]).
--record(buffered_transport, {wrapped, % a thrift_transport
- write_buffer % iolist()
- }).
--type state() :: #buffered_transport{}.
+-record(t_buffered, {
+ wrapped,
+ write_buffer
+}).
+
+-type state() :: #t_buffered{}.
+
+
+-spec new(Transport::thrift_transport:t_transport()) ->
+ thrift_transport:t_transport().
+
+new(Wrapped) ->
+ State = #t_buffered{
+ wrapped = Wrapped,
+ write_buffer = []
+ },
+ thrift_transport:new(?MODULE, State).
+
+
-include("thrift_transport_behaviour.hrl").
-new(WrappedTransport) ->
- State = #buffered_transport{wrapped = WrappedTransport,
- write_buffer = []},
- thrift_transport:new(?MODULE, State).
+%% reads data through from the wrapped transport
+read(State = #t_buffered{wrapped = Wrapped}, Len)
+when is_integer(Len), Len >= 0 ->
+ {NewState, Response} = thrift_transport:read(Wrapped, Len),
+ {State#t_buffered{wrapped = NewState}, Response}.
-%% Writes data into the buffer
-write(State = #buffered_transport{write_buffer = WBuf}, Data) ->
- {State#buffered_transport{write_buffer = [WBuf, Data]}, ok}.
+%% reads data through from the wrapped transport
+read_exact(State = #t_buffered{wrapped = Wrapped}, Len)
+when is_integer(Len), Len >= 0 ->
+ {NewState, Response} = thrift_transport:read_exact(Wrapped, Len),
+ {State#t_buffered{wrapped = NewState}, Response}.
-%% Flushes the buffer through to the wrapped transport
-flush(State = #buffered_transport{write_buffer = WBuf,
- wrapped = Wrapped0}) ->
- {Wrapped1, Response} = thrift_transport:write(Wrapped0, WBuf),
- {Wrapped2, _} = thrift_transport:flush(Wrapped1),
- NewState = State#buffered_transport{write_buffer = [],
- wrapped = Wrapped2},
- {NewState, Response}.
-%% Closes the transport and the wrapped transport
-close(State = #buffered_transport{wrapped = Wrapped0}) ->
- {Wrapped1, Result} = thrift_transport:close(Wrapped0),
- NewState = State#buffered_transport{wrapped = Wrapped1},
- {NewState, Result}.
+write(State = #t_buffered{write_buffer = Buffer}, Data) ->
+ {State#t_buffered{write_buffer = [Buffer, Data]}, ok}.
-%% Reads data through from the wrapped transport
-read(State = #buffered_transport{wrapped = Wrapped0}, Len) when is_integer(Len) ->
- {Wrapped1, Response} = thrift_transport:read(Wrapped0, Len),
- NewState = State#buffered_transport{wrapped = Wrapped1},
- {NewState, Response}.
+
+flush(State = #t_buffered{wrapped = Wrapped, write_buffer = Buffer}) ->
+ case iolist_size(Buffer) of
+ %% if write buffer is empty, do nothing
+ 0 -> {State, ok};
+ _ ->
+ {Written, Response} = thrift_transport:write(Wrapped, Buffer),
+ {Flushed, ok} = thrift_transport:flush(Written),
+ {State#t_buffered{wrapped = Flushed, write_buffer = []}, Response}
+ end.
+
+
+close(State = #t_buffered{wrapped = Wrapped}) ->
+ {Closed, Result} = thrift_transport:close(Wrapped),
+ {State#t_buffered{wrapped = Closed}, Result}.
+
%%--------------------------------------------------------------------
%%% Internal functions
%%--------------------------------------------------------------------
%%%% FACTORY GENERATION %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
new_transport_factory(WrapFactory) ->
- F = fun() ->
- {ok, Wrapped} = WrapFactory(),
- new(Wrapped)
- end,
- {ok, F}.
+ F = fun() ->
+ {ok, Wrapped} = WrapFactory(),
+ new(Wrapped)
+ end,
+ {ok, F}.
+
diff --git a/lib/erl/src/thrift_file_transport.erl b/lib/erl/src/thrift_file_transport.erl
index ba3aa89..071152b 100644
--- a/lib/erl/src/thrift_file_transport.erl
+++ b/lib/erl/src/thrift_file_transport.erl
@@ -21,69 +21,95 @@
-behaviour(thrift_transport).
--export([new_reader/1,
- new/1,
- new/2,
- write/2, read/2, flush/1, close/1]).
+%% constructors
+-export([new/1, new/2]).
+%% protocol callbacks
+-export([read/2, read_exact/2, write/2, flush/1, close/1]).
+%% legacy api
+-export([new_reader/1]).
--record(t_file_transport, {device,
- should_close = true,
- mode = write}).
--type state() :: #t_file_transport{}.
--include("thrift_transport_behaviour.hrl").
-%%%% CONSTRUCTION %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+-record(t_file, {
+ device,
+ should_close = true,
+ mode = write
+}).
-new_reader(Filename) ->
- case file:open(Filename, [read, binary, {read_ahead, 1024*1024}]) of
- {ok, IODevice} ->
- new(IODevice, [{should_close, true}, {mode, read}]);
- Error -> Error
- end.
+-type state() :: #t_file{}.
-new(Device) ->
- new(Device, []).
-%% Device :: io_device()
-%%
+-spec new(Device::file:io_device()) ->
+ thrift_transport:t_transport().
+
+new(Device) -> new(Device, []).
+
+-spec new(Device::file:io_device(), Opts::list()) ->
+ thrift_transport:t_transport().
+
%% Device should be opened in raw and binary mode.
new(Device, Opts) when is_list(Opts) ->
- State = parse_opts(Opts, #t_file_transport{device = Device}),
- thrift_transport:new(?MODULE, State).
+ State = parse_opts(Opts, #t_file{device = Device}),
+ thrift_transport:new(?MODULE, State).
-%% Parse options
-parse_opts([{should_close, Bool} | Rest], State) when is_boolean(Bool) ->
- parse_opts(Rest, State#t_file_transport{should_close = Bool});
-parse_opts([{mode, Mode} | Rest], State)
- when Mode =:= write;
- Mode =:= read ->
- parse_opts(Rest, State#t_file_transport{mode = Mode});
+parse_opts([{should_close, Bool}|Rest], State)
+when is_boolean(Bool) ->
+ parse_opts(Rest, State#t_file{should_close = Bool});
+parse_opts([{mode, Mode}|Rest], State)
+when Mode =:= write; Mode =:= read ->
+ parse_opts(Rest, State#t_file{mode = Mode});
parse_opts([], State) ->
- State.
+ State.
-%%%% TRANSPORT IMPL %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-
-write(This = #t_file_transport{device = Device, mode = write}, Data) ->
- {This, file:write(Device, Data)};
-write(This, _D) ->
- {This, {error, read_mode}}.
+-include("thrift_transport_behaviour.hrl").
-read(This = #t_file_transport{device = Device, mode = read}, Len)
- when is_integer(Len), Len >= 0 ->
- {This, file:read(Device, Len)};
-read(This, _D) ->
- {This, {error, read_mode}}.
+read(State = #t_file{device = Device, mode = read}, Len)
+when is_integer(Len), Len >= 0 ->
+ case file:read(Device, Len) of
+ eof -> {State, {error, eof}};
+ {ok, Result} -> {State, {ok, iolist_to_binary(Result)}}
+ end;
+read(State, _) ->
+ {State, {error, write_mode}}.
-flush(This = #t_file_transport{device = Device, mode = write}) ->
- {This, file:sync(Device)}.
-close(This = #t_file_transport{device = Device, should_close = SC}) ->
- case SC of
- true ->
- {This, file:close(Device)};
- false ->
- {This, ok}
- end.
+read_exact(State = #t_file{device = Device, mode = read}, Len)
+when is_integer(Len), Len >= 0 ->
+ case file:read(Device, Len) of
+ eof -> {State, {error, eof}};
+ {ok, Result} ->
+ case iolist_size(Result) of
+ X when X < Len -> {State, {error, eof}};
+ _ -> {State, {ok, iolist_to_binary(Result)}}
+ end
+ end;
+read_exact(State, _) ->
+ {State, {error, write_mode}}.
+
+
+write(State = #t_file{device = Device, mode = write}, Data) ->
+ {State, file:write(Device, Data)};
+write(State, _) ->
+ {State, {error, read_mode}}.
+
+
+flush(State = #t_file{device = Device, mode = write}) ->
+ {State, file:sync(Device)}.
+
+
+close(State = #t_file{device = Device, should_close = SC}) ->
+ case SC of
+ true -> {State, file:close(Device)};
+ false -> {State, ok}
+ end.
+
+
+%% legacy api. left for compatibility
+new_reader(Filename) ->
+ case file:open(Filename, [read, binary, {read_ahead, 1024*1024}]) of
+ {ok, IODevice} -> new(IODevice, [{should_close, true}, {mode, read}]);
+ Error -> Error
+ end.
+
diff --git a/lib/erl/src/thrift_framed_transport.erl b/lib/erl/src/thrift_framed_transport.erl
index eca3cbe..715f090 100644
--- a/lib/erl/src/thrift_framed_transport.erl
+++ b/lib/erl/src/thrift_framed_transport.erl
@@ -21,83 +21,105 @@
-behaviour(thrift_transport).
-%% API
+%% constructor
-export([new/1]).
+%% protocol callbacks
+-export([read/2, read_exact/2, write/2, flush/1, close/1]).
-%% thrift_transport callbacks
--export([write/2, read/2, flush/1, close/1]).
--record(framed_transport, {wrapped, % a thrift_transport
- read_buffer, % iolist()
- write_buffer % iolist()
- }).
--type state() :: #framed_transport{}.
+-record(t_framed, {
+ wrapped,
+ read_buffer,
+ write_buffer
+}).
+
+-type state() :: #t_framed{}.
+
+
+-spec new(Transport::thrift_transport:t_transport()) ->
+ thrift_transport:t_transport().
+
+new(Wrapped) ->
+ State = #t_framed{
+ wrapped = Wrapped,
+ read_buffer = [],
+ write_buffer = []
+ },
+ thrift_transport:new(?MODULE, State).
+
+
-include("thrift_transport_behaviour.hrl").
-new(WrappedTransport) ->
- State = #framed_transport{wrapped = WrappedTransport,
- read_buffer = [],
- write_buffer = []},
- thrift_transport:new(?MODULE, State).
-%% Writes data into the buffer
-write(State = #framed_transport{write_buffer = WBuf}, Data) ->
- {State#framed_transport{write_buffer = [WBuf, Data]}, ok}.
+read(State = #t_framed{wrapped = Wrapped, read_buffer = Buffer}, Len)
+when is_integer(Len), Len >= 0 ->
+ Binary = iolist_to_binary(Buffer),
+ case Binary of
+ <<>> when Len > 0 ->
+ case next_frame(Wrapped) of
+ {NewState, {ok, Frame}} ->
+ NewBinary = iolist_to_binary([Binary, Frame]),
+ Give = min(iolist_size(NewBinary), Len),
+ {Result, Remaining} = split_binary(NewBinary, Give),
+ {State#t_framed{wrapped = NewState, read_buffer = Remaining}, {ok, Result}};
+ Error -> Error
+ end;
+ %% read of zero bytes
+ <<>> -> {State, {ok, <<>>}};
+ %% read buffer is nonempty
+ _ ->
+ Give = min(iolist_size(Binary), Len),
+ {Result, Remaining} = split_binary(Binary, Give),
+ {State#t_framed{read_buffer = Remaining}, {ok, Result}}
+ end.
-%% Flushes the buffer through to the wrapped transport
-flush(State0 = #framed_transport{write_buffer = Buffer,
- wrapped = Wrapped0}) ->
- FrameLen = iolist_size(Buffer),
- Data = [<<FrameLen:32/integer-signed-big>>, Buffer],
- {Wrapped1, Response} = thrift_transport:write(Wrapped0, Data),
+read_exact(State = #t_framed{wrapped = Wrapped, read_buffer = Buffer}, Len)
+when is_integer(Len), Len >= 0 ->
+ Binary = iolist_to_binary(Buffer),
+ case iolist_size(Binary) of
+ %% read buffer is larger than requested read size
+ X when X >= Len ->
+ {Result, Remaining} = split_binary(Binary, Len),
+ {State#t_framed{read_buffer = Remaining}, {ok, Result}};
+ %% read buffer is insufficient for requested read size
+ _ ->
+ case next_frame(Wrapped) of
+ {NewState, {ok, Frame}} ->
+ read_exact(
+ State#t_framed{wrapped = NewState, read_buffer = [Buffer, Frame]},
+ Len
+ );
+ {NewState, Error} ->
+ {State#t_framed{wrapped = NewState}, Error}
+ end
+ end.
- {Wrapped2, _} = thrift_transport:flush(Wrapped1),
+next_frame(Transport) ->
+ case thrift_transport:read_exact(Transport, 4) of
+ {NewState, {ok, <<FrameLength:32/integer-signed-big>>}} ->
+ thrift_transport:read_exact(NewState, FrameLength);
+ Error -> Error
+ end.
- State1 = State0#framed_transport{wrapped = Wrapped2, write_buffer = []},
- {State1, Response}.
-%% Closes the transport and the wrapped transport
-close(State = #framed_transport{wrapped = Wrapped0}) ->
- {Wrapped1, Result} = thrift_transport:close(Wrapped0),
- NewState = State#framed_transport{wrapped = Wrapped1},
- {NewState, Result}.
+write(State = #t_framed{write_buffer = Buffer}, Data) ->
+ {State#t_framed{write_buffer = [Buffer, Data]}, ok}.
-%% Reads data through from the wrapped transport
-read(State0 = #framed_transport{wrapped = Wrapped0, read_buffer = RBuf},
- Len) when is_integer(Len) ->
- {Wrapped1, {RBuf1, RBuf1Size}} =
- %% if the read buffer is empty, read another frame
- %% otherwise, just read from what's left in the buffer
- case iolist_size(RBuf) of
- 0 ->
- %% read the frame length
- case thrift_transport:read(Wrapped0, 4) of
- {WrappedS1,
- {ok, <<FrameLen:32/integer-signed-big, _/binary>>}} ->
- %% then read the data
- case thrift_transport:read(WrappedS1, FrameLen) of
- {WrappedS2, {ok, Bin}} ->
- {WrappedS2, {Bin, erlang:byte_size(Bin)}};
- {WrappedS2, {error, Reason1}} ->
- {WrappedS2, {error, Reason1}}
- end;
- {WrappedS1, {error, Reason2}} ->
- {WrappedS1, {error, Reason2}}
- end;
- Sz ->
- {Wrapped0, {RBuf, Sz}}
- end,
- %% pull off Give bytes, return them to the user, leave the rest in the buffer
- case RBuf1 of
- error ->
- { State0#framed_transport {wrapped = Wrapped1, read_buffer = [] },
- {RBuf1, RBuf1Size} };
- _ ->
- Give = min(RBuf1Size, Len),
- <<Data:Give/binary, RBuf2/binary>> = iolist_to_binary(RBuf1),
+flush(State = #t_framed{write_buffer = Buffer, wrapped = Wrapped}) ->
+ case iolist_size(Buffer) of
+ %% if write buffer is empty, do nothing
+ 0 -> {State, ok};
+ FrameLen ->
+ Data = [<<FrameLen:32/integer-signed-big>>, Buffer],
+ {Written, Response} = thrift_transport:write(Wrapped, Data),
+ {Flushed, ok} = thrift_transport:flush(Written),
+ {State#t_framed{wrapped = Flushed, write_buffer = []}, Response}
+ end.
- { State0#framed_transport{wrapped = Wrapped1, read_buffer=RBuf2},
- {ok, Data} }
- end.
+
+close(State = #t_framed{wrapped = Wrapped}) ->
+ {Closed, Result} = thrift_transport:close(Wrapped),
+ {State#t_framed{wrapped = Closed}, Result}.
+
diff --git a/lib/erl/src/thrift_membuffer_transport.erl b/lib/erl/src/thrift_membuffer_transport.erl
new file mode 100644
index 0000000..be9acb2
--- /dev/null
+++ b/lib/erl/src/thrift_membuffer_transport.erl
@@ -0,0 +1,83 @@
+%%
+%% Licensed to the Apache Software Foundation (ASF) under one
+%% or more contributor license agreements. See the NOTICE file
+%% distributed with this work for additional information
+%% regarding copyright ownership. The ASF licenses this file
+%% to you under the Apache License, Version 2.0 (the
+%% "License"); you may not use this file except in compliance
+%% with the License. You may obtain a copy of the License at
+%%
+%% http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing,
+%% software distributed under the License is distributed on an
+%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+%% KIND, either express or implied. See the License for the
+%% specific language governing permissions and limitations
+%% under the License.
+%%
+
+-module(thrift_membuffer_transport).
+
+-behaviour(thrift_transport).
+
+%% constructors
+-export([new/0, new/1]).
+%% protocol callbacks
+-export([read/2, read_exact/2, write/2, flush/1, close/1]).
+
+
+-record(t_membuffer, {
+ buffer = []
+}).
+
+-type state() :: #t_membuffer{}.
+
+
+-spec new() -> thrift_transport:t_transport().
+
+new() -> new([]).
+
+-spec new(Buf::iodata()) -> thrift_transport:t_transport().
+
+new(Buf) when is_list(Buf) ->
+ State = #t_membuffer{buffer = Buf},
+ thrift_transport:new(?MODULE, State);
+new(Buf) when is_binary(Buf) ->
+ State = #t_membuffer{buffer = [Buf]},
+ thrift_transport:new(?MODULE, State).
+
+
+-include("thrift_transport_behaviour.hrl").
+
+
+read(State = #t_membuffer{buffer = Buf}, Len)
+when is_integer(Len), Len >= 0 ->
+ Binary = iolist_to_binary(Buf),
+ Give = min(iolist_size(Binary), Len),
+ {Result, Remaining} = split_binary(Binary, Give),
+ {State#t_membuffer{buffer = Remaining}, {ok, Result}}.
+
+
+read_exact(State = #t_membuffer{buffer = Buf}, Len)
+when is_integer(Len), Len >= 0 ->
+ Binary = iolist_to_binary(Buf),
+ case iolist_size(Binary) of
+ X when X >= Len ->
+ {Result, Remaining} = split_binary(Binary, Len),
+ {State#t_membuffer{buffer = Remaining}, {ok, Result}};
+ _ ->
+ {State, {error, eof}}
+ end.
+
+
+write(State = #t_membuffer{buffer = Buf}, Data)
+when is_list(Data); is_binary(Data) ->
+ {State#t_membuffer{buffer = [Buf, Data]}, ok}.
+
+
+flush(State) -> {State, ok}.
+
+
+close(State) -> {State, ok}.
+
diff --git a/lib/erl/src/thrift_memory_buffer.erl b/lib/erl/src/thrift_memory_buffer.erl
index 53abbc4..6a59ea5 100644
--- a/lib/erl/src/thrift_memory_buffer.erl
+++ b/lib/erl/src/thrift_memory_buffer.erl
@@ -21,42 +21,27 @@
-behaviour(thrift_transport).
-%% API
--export([new/0, new/1, new_transport_factory/0]).
+%% constructors
+-export([new/0, new/1]).
+%% protocol callbacks
+-export([read/2, write/2, flush/1, close/1]).
+%% legacy api
+-export([new_transport_factory/0]).
-%% thrift_transport callbacks
--export([write/2, read/2, flush/1, close/1]).
--record(memory_buffer, {buffer}).
--type state() :: #memory_buffer{}.
--include("thrift_transport_behaviour.hrl").
+%% wrapper around thrift_membuffer_transport for legacy reasons
-new() ->
- State = #memory_buffer{buffer = []},
- thrift_transport:new(?MODULE, State).
+new() -> thrift_membuffer_transport:new().
-new (Buf) when is_list (Buf) ->
- State = #memory_buffer{buffer = Buf},
- thrift_transport:new(?MODULE, State);
-new (Buf) ->
- State = #memory_buffer{buffer = [Buf]},
- thrift_transport:new(?MODULE, State).
+new(State) -> thrift_membuffer_transport:new(State).
-new_transport_factory() ->
- {ok, fun() -> new() end}.
+new_transport_factory() -> {ok, fun() -> new() end}.
-%% Writes data into the buffer
-write(State = #memory_buffer{buffer = Buf}, Data) ->
- {State#memory_buffer{buffer = [Buf, Data]}, ok}.
+write(State, Data) -> thrift_membuffer_transport:write(State, Data).
-flush(State = #memory_buffer {buffer = Buf}) ->
- {State#memory_buffer{buffer = []}, Buf}.
+read(State, Data) -> thrift_membuffer_transport:read(State, Data).
-close(State) ->
- {State, ok}.
+flush(State) -> thrift_membuffer_transport:flush(State).
-read(State = #memory_buffer{buffer = Buf}, Len) when is_integer(Len) ->
- Binary = iolist_to_binary(Buf),
- Give = min(iolist_size(Binary), Len),
- {Result, Remaining} = split_binary(Binary, Give),
- {State#memory_buffer{buffer = Remaining}, {ok, Result}}.
+close(State) -> thrift_membuffer_transport:close(State).
+
diff --git a/lib/erl/src/thrift_socket_transport.erl b/lib/erl/src/thrift_socket_transport.erl
index fec0241..fa10ed0 100644
--- a/lib/erl/src/thrift_socket_transport.erl
+++ b/lib/erl/src/thrift_socket_transport.erl
@@ -21,104 +21,156 @@
-behaviour(thrift_transport).
--export([new/1,
- new/2,
- write/2, read/2, flush/1, close/1,
+%% constructors
+-export([new/1, new/2]).
+%% transport callbacks
+-export([read/2, read_exact/2, write/2, flush/1, close/1]).
+%% legacy api
+-export([new_transport_factory/3]).
- new_transport_factory/3]).
--record(data, {socket,
- recv_timeout=infinity}).
--type state() :: #data{}.
--include("thrift_transport_behaviour.hrl").
+-record(t_socket, {
+ socket,
+ recv_timeout=60000,
+ buffer = []
+}).
-new(Socket) ->
- new(Socket, []).
+-type state() :: #t_socket{}.
+
+
+-spec new(Socket::any()) ->
+ thrift_transport:t_transport().
+
+new(Socket) -> new(Socket, []).
+
+-spec new(Socket::any(), Opts::list()) ->
+ thrift_transport:t_transport().
new(Socket, Opts) when is_list(Opts) ->
- State =
- case lists:keysearch(recv_timeout, 1, Opts) of
- {value, {recv_timeout, Timeout}}
- when is_integer(Timeout), Timeout > 0 ->
- #data{socket=Socket, recv_timeout=Timeout};
- _ ->
- #data{socket=Socket}
- end,
- thrift_transport:new(?MODULE, State).
-
-%% Data :: iolist()
-write(This = #data{socket = Socket}, Data) ->
- {This, gen_tcp:send(Socket, Data)}.
-
-read(This = #data{socket=Socket, recv_timeout=Timeout}, Len)
- when is_integer(Len), Len >= 0 ->
- case gen_tcp:recv(Socket, Len, Timeout) of
- Err = {error, timeout} ->
- gen_tcp:close(Socket),
- {This, Err};
- Data ->
- {This, Data}
- end.
-
-%% We can't really flush - everything is flushed when we write
-flush(This) ->
- {This, ok}.
-
-close(This = #data{socket = Socket}) ->
- {This, gen_tcp:close(Socket)}.
+ State = parse_opts(Opts, #t_socket{socket = Socket}),
+ thrift_transport:new(?MODULE, State).
-%%%% FACTORY GENERATION %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+parse_opts([{recv_timeout, Timeout}|Rest], State)
+when is_integer(Timeout), Timeout > 0 ->
+ parse_opts(Rest, State#t_socket{recv_timeout = Timeout});
+parse_opts([{recv_timeout, infinity}|Rest], State) ->
+ parse_opts(Rest, State#t_socket{recv_timeout = infinity});
+parse_opts([], State) ->
+ State.
+-include("thrift_transport_behaviour.hrl").
+
+
+read(State = #t_socket{buffer = Buf}, Len)
+when is_integer(Len), Len >= 0 ->
+ Binary = iolist_to_binary(Buf),
+ case iolist_size(Binary) of
+ X when X >= Len ->
+ {Result, Remaining} = split_binary(Binary, Len),
+ {State#t_socket{buffer = Remaining}, {ok, Result}};
+ _ -> recv(State, Len)
+ end.
+
+recv(State = #t_socket{socket = Socket, buffer = Buf}, Len) ->
+ case gen_tcp:recv(Socket, 0, State#t_socket.recv_timeout) of
+ {error, Error} ->
+ gen_tcp:close(Socket),
+ {State, {error, Error}};
+ {ok, Data} ->
+ Binary = iolist_to_binary([Buf, Data]),
+ Give = min(iolist_size(Binary), Len),
+ {Result, Remaining} = split_binary(Binary, Give),
+ {State#t_socket{buffer = Remaining}, {ok, Result}}
+ end.
+
+
+read_exact(State = #t_socket{buffer = Buf}, Len)
+when is_integer(Len), Len >= 0 ->
+ Binary = iolist_to_binary(Buf),
+ case iolist_size(Binary) of
+ X when X >= Len -> read(State, Len);
+ X ->
+ case gen_tcp:recv(State#t_socket.socket, Len - X, State#t_socket.recv_timeout) of
+ {error, Error} ->
+ gen_tcp:close(State#t_socket.socket),
+ {State, {error, Error}};
+ {ok, Data} ->
+ {State#t_socket{buffer = []}, {ok, <<Binary/binary, Data/binary>>}}
+ end
+ end.
+
+
+write(State = #t_socket{socket = Socket}, Data) ->
+ case gen_tcp:send(Socket, Data) of
+ {error, Error} ->
+ gen_tcp:close(Socket),
+ {State, {error, Error}};
+ ok -> {State, ok}
+ end.
+
+
+flush(State) ->
+ {State#t_socket{buffer = []}, ok}.
+
+
+close(State = #t_socket{socket = Socket}) ->
+ {State, gen_tcp:close(Socket)}.
+
+
+%% legacy api. left for compatibility
+
%% The following "local" record is filled in by parse_factory_options/2
%% below. These options can be passed to new_protocol_factory/3 in a
%% proplists-style option list. They're parsed like this so it is an O(n)
%% operation instead of O(n^2)
--record(factory_opts, {connect_timeout = infinity,
- sockopts = [],
- framed = false}).
+-record(factory_opts, {
+ connect_timeout = infinity,
+ sockopts = [],
+ framed = false
+}).
-parse_factory_options([], Opts) ->
- Opts;
-parse_factory_options([{framed, Bool} | Rest], Opts) when is_boolean(Bool) ->
- parse_factory_options(Rest, Opts#factory_opts{framed=Bool});
-parse_factory_options([{sockopts, OptList} | Rest], Opts) when is_list(OptList) ->
- parse_factory_options(Rest, Opts#factory_opts{sockopts=OptList});
-parse_factory_options([{connect_timeout, TO} | Rest], Opts) when TO =:= infinity; is_integer(TO) ->
- parse_factory_options(Rest, Opts#factory_opts{connect_timeout=TO});
-parse_factory_options([{recv_timeout, TO} | Rest], Opts) when TO =:= infinity; is_integer(TO) ->
- parse_factory_options(Rest, Opts).
+parse_factory_options([], FactoryOpts, TransOpts) -> {FactoryOpts, TransOpts};
+parse_factory_options([{framed, Bool}|Rest], FactoryOpts, TransOpts)
+when is_boolean(Bool) ->
+ parse_factory_options(Rest, FactoryOpts#factory_opts{framed = Bool}, TransOpts);
+parse_factory_options([{sockopts, OptList}|Rest], FactoryOpts, TransOpts)
+when is_list(OptList) ->
+ parse_factory_options(Rest, FactoryOpts#factory_opts{sockopts = OptList}, TransOpts);
+parse_factory_options([{connect_timeout, TO}|Rest], FactoryOpts, TransOpts)
+when TO =:= infinity; is_integer(TO) ->
+ parse_factory_options(Rest, FactoryOpts#factory_opts{connect_timeout = TO}, TransOpts);
+parse_factory_options([{recv_timeout, TO}|Rest], FactoryOpts, TransOpts)
+when TO =:= infinity; is_integer(TO) ->
+ parse_factory_options(Rest, FactoryOpts, [{recv_timeout, TO}] ++ TransOpts).
-%%
%% Generates a "transport factory" function - a fun which returns a thrift_transport()
%% instance.
-%% This can be passed into a protocol factory to generate a connection to a
+%% State can be passed into a protocol factory to generate a connection to a
%% thrift server over a socket.
-%%
new_transport_factory(Host, Port, Options) ->
- ParsedOpts = parse_factory_options(Options, #factory_opts{}),
-
- F = fun() ->
- SockOpts = [binary,
- {packet, 0},
- {active, false},
- {nodelay, true} |
- ParsedOpts#factory_opts.sockopts],
- case catch gen_tcp:connect(Host, Port, SockOpts,
- ParsedOpts#factory_opts.connect_timeout) of
- {ok, Sock} ->
- {ok, Transport} =
- thrift_socket_transport:new(Sock, Options),
- {ok, BufTransport} =
- case ParsedOpts#factory_opts.framed of
- true -> thrift_framed_transport:new(Transport);
- false -> thrift_buffered_transport:new(Transport)
- end,
- {ok, BufTransport};
- Error ->
- Error
- end
+ {FactoryOpts, TransOpts} = parse_factory_options(Options, #factory_opts{}, []),
+ {ok, fun() -> SockOpts = [binary,
+ {packet, 0},
+ {active, false},
+ {nodelay, true}|FactoryOpts#factory_opts.sockopts
+ ],
+ case catch gen_tcp:connect(
+ Host,
+ Port,
+ SockOpts,
+ FactoryOpts#factory_opts.connect_timeout
+ ) of
+ {ok, Sock} ->
+ {ok, Transport} = thrift_socket_transport:new(Sock, TransOpts),
+ {ok, BufTransport} = case FactoryOpts#factory_opts.framed of
+ true -> thrift_framed_transport:new(Transport);
+ false -> thrift_buffered_transport:new(Transport)
end,
- {ok, F}.
+ {ok, BufTransport};
+ Error -> Error
+ end
+ end}.
+
diff --git a/lib/erl/src/thrift_transport.erl b/lib/erl/src/thrift_transport.erl
index 39f8c05..0fdf970 100644
--- a/lib/erl/src/thrift_transport.erl
+++ b/lib/erl/src/thrift_transport.erl
@@ -20,59 +20,102 @@
-module(thrift_transport).
-export([behaviour_info/1]).
+%% constructors
+-export([new/1, new/2]).
+%% transport callbacks
+-export([read/2, read_exact/2, write/2, flush/1, close/1]).
--export([new/2,
- write/2,
- read/2,
- flush/1,
- close/1
- ]).
+-export_type([t_transport/0]).
+
behaviour_info(callbacks) ->
- [{read, 2},
- {write, 2},
- {flush, 1},
- {close, 1}
- ].
+ [{read, 2}, {write, 2}, {flush, 1}, {close, 1}].
--record(transport, {module, data}).
+
+-record(t_transport, {
+ module,
+ state
+}).
+
+-type state() :: #t_transport{}.
+-type t_transport() :: #t_transport{}.
+
-ifdef(transport_wrapper_module).
-define(debug_wrap(Transport),
- case Transport#transport.module of
- ?transport_wrapper_module ->
- Transport;
- _Else ->
- {ok, Result} = ?transport_wrapper_module:new(Transport),
- Result
- end).
+ case Transport#t_transport.module of
+ ?transport_wrapper_module -> Transport;
+ _Else ->
+ {ok, Result} = ?transport_wrapper_module:new(Transport),
+ Result
+ end
+).
-else.
-define(debug_wrap(Transport), Transport).
-endif.
-new(Module, Data) when is_atom(Module) ->
- Transport0 = #transport{module = Module, data = Data},
- Transport1 = ?debug_wrap(Transport0),
- {ok, Transport1}.
--spec write(#transport{}, iolist() | binary()) -> {#transport{}, ok | {error, _Reason}}.
-write(Transport, Data) ->
- Module = Transport#transport.module,
- {NewTransData, Result} = Module:write(Transport#transport.data, Data),
- {Transport#transport{data = NewTransData}, Result}.
+-type wrappable() ::
+ binary() |
+ list() |
+ {membuffer, binary() | list()} |
+ {tcp, port()} |
+ {tcp, port(), list()} |
+ {file, file:io_device()} |
+ {file, file:io_device(), list()} |
+ {file, file:filename()} |
+ {file, file:filename(), list()}.
--spec read(#transport{}, non_neg_integer()) -> {#transport{}, {ok, binary()} | {error, _Reason}}.
-read(Transport, Len) when is_integer(Len) ->
- Module = Transport#transport.module,
- {NewTransData, Result} = Module:read(Transport#transport.data, Len),
- {Transport#transport{data = NewTransData}, Result}.
+-spec new(wrappable()) -> {ok, #t_transport{}}.
--spec flush(#transport{}) -> {#transport{}, ok | {error, _Reason}}.
-flush(Transport = #transport{module = Module, data = Data}) ->
- {NewTransData, Result} = Module:flush(Data),
- {Transport#transport{data = NewTransData}, Result}.
+new({membuffer, Membuffer}) when is_binary(Membuffer); is_list(Membuffer) ->
+ thrift_membuffer_transport:new(Membuffer);
+new({membuffer, Membuffer, []}) when is_binary(Membuffer); is_list(Membuffer) ->
+ thrift_membuffer_transport:new(Membuffer);
+new({tcp, Socket}) when is_port(Socket) ->
+ new({tcp, Socket, []});
+new({tcp, Socket, Opts}) when is_port(Socket) ->
+ thrift_socket_transport:new(Socket, Opts);
+new({file, Filename}) when is_list(Filename); is_binary(Filename) ->
+ new({file, Filename, []});
+new({file, Filename, Opts}) when is_list(Filename); is_binary(Filename) ->
+ {ok, File} = file:open(Filename, [raw, binary]),
+ new({file, File, Opts});
+new({file, File, Opts}) ->
+ thrift_file_transport:new(File, Opts).
--spec close(#transport{}) -> {#transport{}, ok | {error, _Reason}}.
-close(Transport = #transport{module = Module, data = Data}) ->
- {NewTransData, Result} = Module:close(Data),
- {Transport#transport{data = NewTransData}, Result}.
+-spec new(Module::module(), State::any()) -> {ok, #t_transport{}}.
+
+new(Module, State) when is_atom(Module) ->
+ {ok, ?debug_wrap(#t_transport{module = Module, state = State})}.
+
+
+-include("thrift_transport_behaviour.hrl").
+
+
+read(Transport = #t_transport{module = Module}, Len)
+when is_integer(Len), Len >= 0 ->
+ {NewState, Result} = Module:read(Transport#t_transport.state, Len),
+ {Transport#t_transport{state = NewState}, Result}.
+
+
+read_exact(Transport = #t_transport{module = Module}, Len)
+when is_integer(Len), Len >= 0 ->
+ {NewState, Result} = Module:read_exact(Transport#t_transport.state, Len),
+ {Transport#t_transport{state = NewState}, Result}.
+
+
+write(Transport = #t_transport{module = Module}, Data) ->
+ {NewState, Result} = Module:write(Transport#t_transport.state, Data),
+ {Transport#t_transport{state = NewState}, Result}.
+
+
+flush(Transport = #t_transport{module = Module}) ->
+ {NewState, Result} = Module:flush(Transport#t_transport.state),
+ {Transport#t_transport{state = NewState}, Result}.
+
+
+close(Transport = #t_transport{module = Module}) ->
+ {NewState, Result} = Module:close(Transport#t_transport.state),
+ {Transport#t_transport{state = NewState}, Result}.
+