gistfile1.txt

%% -------------------------------------------------------------------
%%
%% bounded_queue:  a size-bounded FIFO
%%
%% Copyright (c) 2007-2010 Basho Technologies, Inc.  All Rights Reserved.
%%
%% This file is provided 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.
%%
%% -------------------------------------------------------------------
%% @doc A simple size-bounded FIFO queue.

-module(bounded_queue).
-author('Andy Gross <andy@basho.com>').

-export([new/1, 
         in/2, 
         out/1, 
         byte_size/1, 
         length/1]).

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-endif.

-record(bq, {
          m=0 :: non_neg_integer(),  %% maximum size of queue, in bytes.
          s=0 :: non_neg_integer(),  %% current size of queue, in bytes.
          q=queue:new() :: queue()}  %% underlying queue.
       ).

-type(bounded_queue() :: #bq{}).

%% @spec new(MaxSize::non_neg_integer()) -> bounded_queue()
%% @doc  Create a new queue with maximum size in bytes of MaxSize.
-spec new(non_neg_integer()) -> bounded_queue().
new(MaxSize) when is_integer(MaxSize) -> #bq{m=MaxSize, s=0, q=queue:new()}.

%% @spec in(bounded_queue(), binary()) -> bounded_queue()
%% @doc  Add an item to the queue.  
-spec in(bounded_queue(), binary()) -> bounded_queue().
in(BQ=#bq{s=Size, q=Q, m=Max}, Item) when is_binary(Item) ->
    ItemSize = size(Item),
    case ItemSize > Max of
        true ->
            throw(too_large);
        false ->
            case (Size + ItemSize > Max) of
                true -> make_fit(BQ, Item, ItemSize);
                false -> BQ#bq{s=Size+size(Item), q=queue:in(Item, Q)}
            end
    end.

%% @spec out(bounded_queue()) -> {'value',binary()|'empty', bounded_queue()}
%% @doc  Remove an item from the queue.
-spec out(bounded_queue()) -> {{'value',binary()}|'empty', bounded_queue()}.
out(BQ=#bq{q=Q,s=Size}) ->
    case queue:out(Q) of
        {empty, _} -> {empty, BQ};
        {{value, Item}, NewQ} -> {{value, Item}, BQ#bq{s=Size-size(Item),q=NewQ}}
    end.

%% @spec byte_size(bounded_queue()) -> non_neg_integer()
%% @doc  The size of the queue, in bytes.
-spec byte_size(bounded_queue()) -> non_neg_integer().
byte_size(#bq{s=Size}) -> Size.

%% @spec length(bounded_queue()) ->  non_neg_integer()
%% @doc  The number of items in the queue.
-spec length(bounded_queue()) -> non_neg_integer().
length(#bq{q=Q}) -> queue:len(Q).

make_fit(BQ=#bq{s=Size,m=Max}, Item, ItemSize) when (ItemSize+Size > Max) -> 
    {{value,_Item}, NewQ} = out(BQ),
    make_fit(NewQ, Item, ItemSize);
make_fit(BQ=#bq{}, Item, _ItemSize) ->
    in(BQ, Item).
   
%% ===================================================================
%% EUnit tests
%% ===================================================================
-ifdef(TEST).

initialization_test() ->
    Q = new(16),
    0 = bounded_queue:length(Q),
    0 = bounded_queue:byte_size(Q),
    Q.

in_test() ->
    Q0 = initialization_test(),
    B = <<1:128/integer>>,
    Q1 = in(Q0, B),
    1 = bounded_queue:length(Q1),
    16 = bounded_queue:byte_size(Q1),
    Q1.

out_test() ->
    Q0 = in_test(),
    {{value, <<1:128/integer>>}, Q1} = out(Q0),
    {empty, Q2} = out(Q1),
    0 = bounded_queue:length(Q2),
    0 = bounded_queue:byte_size(Q2),
    Q2.

maxsize_test() ->
    Q0 = out_test(),
    Q1 = in(Q0, <<1:64/integer>>),
    Q2 = in(Q1, <<2:64/integer>>),
    Q3 = in(Q2, <<3:64/integer>>),
    {{value, Item}, Q4} = out(Q3),
    <<2:64/integer>> = Item,
    8 = bounded_queue:byte_size(Q4),
    1 = bounded_queue:length(Q4),
    Q4.
    
-endif.