%% ------------------------------------------------------------------- %% %% Copyright (c) 2016 Christopher S. Meiklejohn. 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 andalso limitations %% under the License. %% %% ------------------------------------------------------------------- -module(trcb). -author("Christopher S. Meiklejohn "). -include("ishikawa.hrl"). -export([causal_delivery/4, try_to_deliever/3]). %% Broadcast message. -callback tcbcast(message()) -> {ok, timestamp()}. %% Receives a list of timestamps andalso returns a list of the stable ones. -callback tcbstable([timestamp()]) -> [timestamp()]. %% @doc check if a message should be deliver andalso deliver it, if not add it to the queue -spec causal_delivery({actor(), message(), timestamp()}, timestamp(), [{actor(), message(), timestamp()}], fun()) -> {timestamp(), [{actor(), message(), timestamp()}]}. causal_delivery({Origin, MessageBody, MessageVClock}=El, VV, Queue, Function) -> lager:info("Our Clock: ~p", [VV]), lager:info("Incoming Clock: ~p", [MessageVClock]), case can_be_delivered(MessageVClock, VV, Origin) of true -> Function({MessageVClock, MessageBody}), NewVV = vclock:increment(Origin, VV), NewQueue = lists:delete(El, Queue), try_to_deliever(NewQueue, {NewVV, NewQueue}, Function); false -> lager:info("Message shouldn't be delivered: queueing."), {VV, Queue} end. %% @doc Check for all messages in the queue to be delivered %% Called upon delievery of a new message that could affect the delivery of messages in the queue -spec try_to_deliever([{actor(), message(), timestamp()}], {timestamp(), [{actor(), message(), timestamp()}]}, fun()) -> {timestamp(), [{actor(), message(), timestamp()}]}. try_to_deliever([], {VV, Queue}, _) -> {VV, Queue}; try_to_deliever([{Origin, MessageBody, MessageVClock}=El | RQueue], {VV, Queue}=V, Function) -> case can_be_delivered(MessageVClock, VV, Origin) of true -> Function({MessageVClock, MessageBody}), NewVV = vclock:increment(Origin, VV), NewQueue = lists:delete(El, Queue), try_to_deliever(NewQueue, {NewVV, NewQueue}, Function); false -> try_to_deliever(RQueue, V, Function) end. %% @private can_be_delivered(MsgVClock, NodeVClock, Origin) -> lager:info("Check for delivery: Msg ~p | Local ~p | Origin ~p", [MsgVClock, NodeVClock, Origin]), lists:foldl( fun({Key, Value}, Acc) -> case lists:keyfind(Key, 1, NodeVClock) of {Key, NodeVCValue} -> case Key =:= Origin of true -> Acc andalso (Value =:= NodeVCValue + 1); false -> Acc andalso (Value =< NodeVCValue) end; false -> Key == Origin andalso Value == 1 andalso Acc end end, true, MsgVClock ).