ceph/src/jaegertracing/thrift/lib/erl/src/thrift_framed_transport.erl

   1 %%
   2 %% Licensed to the Apache Software Foundation (ASF) under one
   3 %% or more contributor license agreements. See the NOTICE file
   4 %% distributed with this work for additional information
   5 %% regarding copyright ownership. The ASF licenses this file
   6 %% to you under the Apache License, Version 2.0 (the
   7 %% "License"); you may not use this file except in compliance
   8 %% with the License. You may obtain a copy of the License at
   9 %%
  10 %%   http://www.apache.org/licenses/LICENSE-2.0
  11 %%
  12 %% Unless required by applicable law or agreed to in writing,
  13 %% software distributed under the License is distributed on an
  14 %% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  15 %% KIND, either express or implied. See the License for the
  16 %% specific language governing permissions and limitations
  17 %% under the License.
  18 %%
  19
  20 -module(thrift_framed_transport).
  21
  22 -behaviour(thrift_transport).
  23
  24 %% constructor
  25 -export([new/1]).
  26 %% protocol callbacks
  27 -export([read/2, read_exact/2, write/2, flush/1, close/1]).
  28
  29
  30 -record(t_framed, {
  31   wrapped,
  32   read_buffer,
  33   write_buffer
  34 }).
  35
  36 -type state() :: #t_framed{}.
  37
  38
  39 -spec new(Transport::thrift_transport:t_transport()) ->
  40   thrift_transport:t_transport().
  41
  42 new(Wrapped) ->
  43   State = #t_framed{
  44     wrapped = Wrapped,
  45     read_buffer = [],
  46     write_buffer = []
  47   },
  48   thrift_transport:new(?MODULE, State).
  49
  50
  51 -include("thrift_transport_behaviour.hrl").
  52
  53
  54 read(State = #t_framed{wrapped = Wrapped, read_buffer = Buffer}, Len)
  55 when is_integer(Len), Len >= 0 ->
  56   Binary = iolist_to_binary(Buffer),
  57   case Binary of
  58     <<>> when Len > 0 ->
  59       case next_frame(Wrapped) of
  60         {NewState, {ok, Frame}} ->
  61           NewBinary = iolist_to_binary([Binary, Frame]),
  62           Give = min(iolist_size(NewBinary), Len),
  63           {Result, Remaining} = split_binary(NewBinary, Give),
  64           {State#t_framed{wrapped = NewState, read_buffer = Remaining}, {ok, Result}};
  65         {NewState, Error} ->
  66           {State#t_framed{wrapped = NewState}, Error}
  67       end;
  68     %% read of zero bytes
  69     <<>> -> {State, {ok, <<>>}};
  70     %% read buffer is nonempty
  71     _ ->
  72       Give = min(iolist_size(Binary), Len),
  73       {Result, Remaining} = split_binary(Binary, Give),
  74       {State#t_framed{read_buffer = Remaining}, {ok, Result}}
  75   end.
  76
  77
  78 read_exact(State = #t_framed{wrapped = Wrapped, read_buffer = Buffer}, Len)
  79 when is_integer(Len), Len >= 0 ->
  80   Binary = iolist_to_binary(Buffer),
  81   case iolist_size(Binary) of
  82     %% read buffer is larger than requested read size
  83     X when X >= Len ->
  84       {Result, Remaining} = split_binary(Binary, Len),
  85       {State#t_framed{read_buffer = Remaining}, {ok, Result}};
  86     %% read buffer is insufficient for requested read size
  87     _ ->
  88       case next_frame(Wrapped) of
  89         {NewState, {ok, Frame}} ->
  90           read_exact(
  91             State#t_framed{wrapped = NewState, read_buffer = [Buffer, Frame]},
  92             Len
  93           );
  94         {NewState, Error} ->
  95           {State#t_framed{wrapped = NewState}, Error}
  96       end
  97   end.
  98
  99 next_frame(Transport) ->
 100   case thrift_transport:read_exact(Transport, 4) of
 101     {NewState, {ok, <<FrameLength:32/integer-signed-big>>}} ->
 102       thrift_transport:read_exact(NewState, FrameLength);
 103     Error -> Error
 104   end.
 105
 106
 107 write(State = #t_framed{write_buffer = Buffer}, Data) ->
 108   {State#t_framed{write_buffer = [Buffer, Data]}, ok}.
 109
 110
 111 flush(State = #t_framed{write_buffer = Buffer, wrapped = Wrapped}) ->
 112   case iolist_size(Buffer) of
 113     %% if write buffer is empty, do nothing
 114     0 -> {State, ok};
 115     FrameLen ->
 116       Data = [<<FrameLen:32/integer-signed-big>>, Buffer],
 117       {Written, Response} = thrift_transport:write(Wrapped, Data),
 118       {Flushed, ok} = thrift_transport:flush(Written),
 119       {State#t_framed{wrapped = Flushed, write_buffer = []}, Response}
 120   end.
 121
 122
 123 close(State = #t_framed{wrapped = Wrapped}) ->
 124   {Closed, Result} = thrift_transport:close(Wrapped),
 125   {State#t_framed{wrapped = Closed}, Result}.
 126