diff --git a/apps/roster/src/protocol/roster_schedule.erl b/apps/roster/src/protocol/roster_schedule.erl index 9ac200be836ec8df7768c945af8fa19e33a04022..d1469ed5cd5593bce955909d9009b929d6e7c320 100644 --- a/apps/roster/src/protocol/roster_schedule.erl +++ b/apps/roster/src/protocol/roster_schedule.erl @@ -36,9 +36,13 @@ info(#'History'{feed = {act, <<"publish">>, PhoneId} = Feed, size = N0, end, {reply, {bert, Reply}, Req, State}; -info(#'Job'{id = [], feed_id = {act, <<"publish">>, PhoneId} = Feed, time = Time, +info(#'Job'{id = [], feed_id = {act, <<"publish">>, PhoneId} = Feed, time = Time0, data = [#'Message'{} | _], status = init} = J0, Req, #cx{params = ClientId, client_pid = C} = S) -> - ?ROSTER_LOG_REQ('Job', init, ClientId, "Feed=~p, Time=~p", [Feed, Time]), + ?ROSTER_LOG_REQ('Job', init, ClientId, "Feed=~p, Time=~p", [Feed, Time0]), + Time = case Time0 of + [] -> os:system_time(millisecond); + T when is_integer(T) -> T + end, JId = kvs:next_id('Job', 1), J1 = J0#'Job'{id = JId, status = pending}, Writer = case kvs_stream:load_writer(Feed) of @@ -147,8 +151,8 @@ do_scheduled_work({_TimeStamp, JobId} = SchedKey, MqttClient) -> {aborted, Reason} -> ?LOG_INFO("Worker for ~p aborted, reason: ~p", [JobId, Reason]) end. -do_work(MqttClient, #'Job'{feed_id = #act{data = _PhoneId}, data = [#'Message'{ } = Msg]}) -> - send_msg(MqttClient, Msg). +do_work(MqttClient, #'Job'{feed_id = #act{data = _PhoneId}, data = [#'Message'{} | _] = Msgs}) -> + [send_msg(MqttClient, Msg) || Msg = #'Message'{} <- Msgs]. complete_job(MqttClient, Job = #'Job'{feed_id = #act{data = PhoneId}}) -> Job1 = Job#'Job'{status = complete, data = [{done, os:system_time(millisecond)}]}, @@ -214,6 +218,7 @@ job_history(#'History'{feed = Feed, size = N0, entity_id = JId0} = History) -> Jobs = get_job_history(JId, N), History#'History'{data = Jobs, size = length(Jobs)}. +get_job_history(0, _) -> []; get_job_history(JId, N) -> {Dir, Limit} = if N < 0 -> {bwd, -N}; true -> {fwd, N} end, FoldFun =