Skip to content

Commit d8c31b1

Browse files
gxlzlihaogxlzlihao
committed
add code to record the timer reference in poller
1 parent 8b20200 commit d8c31b1

9 files changed

Lines changed: 54 additions & 38 deletions

File tree

‎include/poller.hrl‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
name,
55
uri,
66
frequency,
7-
pid}).
7+
pid,
8+
timer_ref}).
89

910

‎scripts/python/cgi-bin/humidity.py‎

100644100755
Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@
33
import random
44
timestamp = datetime.datetime.now()
55
humidity = random.random()
6-
temperature = random.random() * 100
76
print "Content-Type: application/json"
87
print
98
print """\
@@ -13,4 +12,4 @@
1312
"humidity": {"value": %f, "timestamp": "%s"}
1413
}
1514
}
16-
""" % (temperature, timestamp.strftime("%Y-%m-%d %H:%M:%S"), humidity, timestamp.strftime("%Y-%m-%d %H:%M:%S"))
15+
""" % (humidity, timestamp.strftime("%Y-%m-%d %H:%M:%S"))

‎scripts/python/cgi-bin/temperature.py‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22
import datetime
33
import random
44
timestamp = datetime.datetime.now()
5-
humidity = random.random()
65
temperature = random.random() * 100
76
print "Content-Type: application/json"
87
print
@@ -13,4 +12,4 @@
1312
"temperature": {"value": %f, "timestamp": "%s"}
1413
}
1514
}
16-
""" % (temperature, timestamp.strftime("%Y-%m-%d %H:%M:%S"), humidity, timestamp.strftime("%Y-%m-%d %H:%M:%S"))
15+
""" % (temperature, timestamp.strftime("%Y-%m-%d %H:%M:%S"))

‎src/parser.erl‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -40,21 +40,21 @@ parseJson(Parser, Data) ->
4040
%%insert the data as a new datapoint into the database
4141
%%since we need more time to investigate how to handle timestamp
4242
%%so currently we only consider the time when we receive the datapackage
43-
case is_integer(StreamId) of
43+
FieldValue1 = case is_integer(StreamId) of
44+
true->
45+
{"stream_id",integer_to_binary(StreamId)};
46+
_ ->
47+
{"stream_id",list_to_binary(StreamId)}
48+
end,
49+
FieldValue2 = case is_integer(Res) of
4450
true->
45-
FieldValue1 = {"stream_id",integer_to_binary(StreamId)};
46-
_ ->
47-
FieldValue1 = {"stream_id",list_to_binary(StreamId)}
48-
end,
49-
case is_integer(Res) of
50-
true->
51-
FieldValue2 = {"value",integer_to_binary(Res)};
51+
{"value",integer_to_binary(Res)};
5252
_->
5353
case is_float(Res) of
5454
true->
55-
FieldValue2 = {"value",float_to_binary(Res)};
55+
{"value",float_to_binary(Res)};
5656
_ ->
57-
FieldValue2 = {"value",list_to_binary(Res)}
57+
{"value",list_to_binary(Res)}
5858
end
5959
end,
6060
{{Year, Month, Day}, {Hour, Minutes, Seconds}} = calendar:now_to_universal_time(os:timestamp()),

‎src/poller.erl‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -68,14 +68,15 @@ handle_call({rebuild}, _Form, State)->
6868
{reply, {error, ErrMsg}, State};
6969
_ ->
7070
NewUri = lib_json:get_field(FinalJson, "uri"),
71+
NewFreq = lib_json:get_field(FinalJson, "polling_freq"),
7172
case is_binary(NewUri) of
7273
false->
7374
FinalUri = NewUri;
7475
_ ->
7576
FinalUri = binary_to_list(NewUri)
7677
end,
7778
%% notify the supervisor to refresh its records
78-
{reply, {update, StreamId, FinalUri}, #state{stream_id=StreamId, uri=FinalUri, parser=Parser}}
79+
{reply, {update, StreamId, FinalUri, NewFreq}, #state{stream_id=StreamId, uri=FinalUri, parser=Parser}}
7980
end
8081
end;
8182
handle_call({check_info}, _Form, State)->

‎src/pollingSystem.erl‎

Lines changed: 14 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -55,8 +55,7 @@ init(_)->
5555
%% Function: handle_info/2
5656
%% Purpose: handle the messages` processing of the gen_server, accepts two parameters: message and the old state of gen_server.
5757
%% could be called by: polling_supervisor!{message}
58-
%% Returns: {noreply, NewState}
59-
%% Side effects: send message to specific poller and update the state of gen_server.
58+
%% Returns: {noreply, State}
6059
%% @end
6160
-spec handle_info(any(), any()) -> tuple().
6261
handle_info({print, Message}, State)->
@@ -77,10 +76,10 @@ handle_cast({create_poller, #pollerInfo{stream_id = StreamId, name = StreamName,
7776
Parser = poll_help:get_parser_by_id(StreamId),
7877

7978
{ok, Pid}=supervisor:start_child(polling_monitor, [#state{stream_id=StreamId, uri=Uri, parser=Parser}]),
80-
Record = #pollerInfo{stream_id=StreamId, name=StreamName, uri=Uri, frequency=Frequency, pid=Pid},
8179
%%use timer library to create scheduler for this poller
8280
timer:start(),
83-
timer:send_interval(Frequency, Pid, {probe}),
81+
{ok, TRef} = timer:send_interval(Frequency, Pid, {probe}),
82+
Record = #pollerInfo{stream_id=StreamId, name=StreamName, uri=Uri, frequency=Frequency, pid=Pid, timer_ref=TRef},
8483
{noreply, [Record|PollersInfo]};
8584
handle_cast({terminate, StreamId}, PollersInfo)->
8685
%% erlang:display("receive cast: {terminate, ResourceId}"),
@@ -103,17 +102,13 @@ handle_cast({rebuild, StreamId}, PollersInfo)->
103102
erlang:display("the error in finding poller: " ++ ErrMessage),
104103
NewPollersInfo = PollersInfo;
105104
_ ->
106-
{update, StreamId, NewUri} = gen_server:call(Poller#pollerInfo.pid, {rebuild}),
105+
{update, StreamId, NewUri, NewFreq} = gen_server:call(Poller#pollerInfo.pid, {rebuild}),
106+
{ok, cancel} = timer:cancel(Poller#pollerInfo.timer_ref),
107+
{ok, NewTRef} = timer:send_interval(NewFreq, Poller#pollerInfo.pid, {probe}),
107108
%%change the url of the poller information stored
108-
NewPollersInfo = update_info(PollersInfo, StreamId, NewUri)
109+
NewPollersInfo = update_info(PollersInfo, StreamId, NewUri, NewTRef)
109110
end,
110-
{noreply, NewPollersInfo};
111-
handle_cast({add_new_poller, Poller}, PollersInfo)->
112-
%% erlang:display("receive cast: {add_new_poller, Poller}"),
113-
{noreply, [Poller|PollersInfo]};
114-
handle_cast({update, StreamId, NewUri}, PollersInfo)->
115-
%% erlang:display("receive cast: {update, ResourceId, NewUrl}"),
116-
{noreply, update_info(PollersInfo, StreamId, NewUri)}.
111+
{noreply, NewPollersInfo}.
117112

118113
%% @doc
119114
%% Function: handle_call/2
@@ -186,17 +181,17 @@ delete_info([Poller|Tail], StreamId)->
186181
end.
187182

188183
%% @doc
189-
%% Function: update_info/3
184+
%% Function: update_info/4
190185
%% Purpose: after poller done its rebuild, supervisor uses this function to update its pollers` info store.
191186
%% Returns: NewPollersRecordList | []
192187
%% @end
193-
-spec update_info(list(tuple()), integer(), string()) -> list().
194-
update_info([], _, _) -> [];
195-
update_info([Poller|Tail], StreamId, NewUri) ->
188+
-spec update_info(list(tuple()), integer(), string(), integer()) -> list().
189+
update_info([], _, _, _) -> [];
190+
update_info([Poller|Tail], StreamId, NewUri, NewTRef) ->
196191
case Poller#pollerInfo.stream_id == StreamId of
197192
true->
198-
[Poller#pollerInfo{uri = NewUri} | Tail];
193+
[Poller#pollerInfo{uri = NewUri, timer_ref = NewTRef} | Tail];
199194
_ ->
200-
[Poller | update_info(Tail, StreamId, NewUri)]
195+
[Poller | update_info(Tail, StreamId, NewUri, NewTRef)]
201196
end.
202197

‎src/suggest.erl‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -212,6 +212,10 @@ add_stream_suggestion_fields(Stream) ->
212212
%% @end
213213
-spec update_suggestion(Stream::json()) -> ok.
214214
update_suggestion(Stream) ->
215+
216+
%% only for testing
217+
erlang:display("run function: update_suggestion/1"),
218+
215219
ResourceId = lib_json:get_field(Stream, "resource_id"),
216220
case erlastic_search:search(?INDEX, "suggestion", "resource_id:"++ lib_json:to_string(ResourceId)) of
217221
{error, _} -> erlang:display("ERROR");
@@ -238,7 +242,9 @@ update_suggestion(Stream) ->
238242
Final = api_help:create_update(NewSugg),
239243
case api_help:update_doc(?INDEX, "suggestion", Id, Final) of
240244
{error, _Reason} -> erlang:display("not updated");
241-
{ok, _Json} -> ok
245+
{ok, _Json} ->
246+
erlang:display("succeed updating the suggestion!!"),
247+
ok
242248
end;
243249
_ ->
244250
erlang:display("No suggestion exists for that resource")

‎test/pollingSystem_tests.erl‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -148,11 +148,15 @@ rebuild_system_test()->
148148

149149
%% testing rebuild
150150
clear_stream_type(),
151+
clear_parser_type(),
151152
api_help:refresh(),
152153

153154
post_stream_with_id(1, "test2", ?POLL_ADD2, 1000, "application/json"),
154155
post_stream_with_id(2, "test1", ?POLL_ADD, 1300, "application/json"),
155156

157+
post_parser(2, "application/json","streams/temperature/value"),
158+
post_parser(1, "application/json","streams/humidity/value"),
159+
156160
api_help:refresh(),
157161
gen_server:cast(polling_supervisor, {rebuild, "1"}),
158162
gen_server:cast(polling_supervisor, {rebuild, "2"}),
@@ -190,12 +194,18 @@ rebuild_system_test()->
190194
erlang:display("!!!!!!!!!!!!!!!!!"),
191195

192196
clear_stream_type(),
197+
clear_parser_type(),
193198
api_help:refresh(),
194199
post_stream_with_id(1, "test", ?POLL_ADD, 1000, "application/json"),
195200
post_stream_with_id(2, "test2", ?POLL_ADD2, 1300, "application/json"),
201+
202+
post_parser(1, "application/json","streams/temperature/value"),
203+
post_parser(2, "application/json","streams/humidity/value"),
204+
196205
api_help:refresh(),
197206
gen_server:cast(polling_supervisor, {rebuild, "1"}),
198207
gen_server:cast(polling_supervisor, {rebuild, "2"}),
208+
timer:sleep(1000),
199209

200210
ChildrenList2 = supervisor:which_children(polling_monitor),
201211
?assertEqual(2, length(ChildrenList2)),

‎test/suggest_tests.erl‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -178,13 +178,18 @@ text_autocompletion_test() ->
178178
{ok, {{_Version, 200, _ReasonPhrase}, _Headers, Body}} = Response1,
179179
Id = lib_json:get_field(Body, "_id"),
180180
api_help:refresh(),
181-
{ok, {{_Version1, 200, _ReasonPhrase1}, _Headers1, Body1}} = httpc:request(post, {"http://localhost:8000/streams", [],"application/json", "{\"name\" : \"search\",\"resource_id\" : \""++lib_json:to_string(Id)++"\", \"private\" : \"false\", \"tags\":\"test_auto_tag\"}"}, [], []),
181+
{ok, {{_Version0, 200, _ReasonPhrase0}, _Headers0, Body0}} = httpc:request(post, {"http://localhost:8000/streams", [],"application/json", "{\"name\" : \"search\",\"resource_id\" : \""++lib_json:to_string(Id)++"\", \"private\" : \"false\", \"tags\":\"test_auto_tag\"}"}, [], []),
182182
api_help:refresh(),
183+
184+
%% os:cmd("lsof | wc -"),
185+
186+
{ok, {{_Version1, 200, _ReasonPhrase1}, _Headers1, Body1}} = httpc:request(post, {"http://localhost:8000/streams", [],"application/json", "{\"name\" : \"search\",\"resource_id\" : \""++lib_json:to_string(Id)++"\", \"private\" : \"false\", \"tags\":\"test_auto_tag\"}"}, [], []),
187+
timer:sleep(1000),
183188
Response2 = get_request(?SUGGEST_URL++"model/"++"testauto"),
184189
Response3 = get_request(?SUGGEST_URL++"tags/"++"testautotag"),
185190
Response4 = get_request(?SUGGEST_URL++"tags/"++"test_auto_tag"),
186-
check_returned_code(Response2, 200),
187191
check_returned_code(Response3, 200),
192+
check_returned_code(Response2, 200),
188193
check_returned_code(Response4, 200),
189194
{ok, {_, _ ,Body2}} = Response2,
190195
{ok, {_, _ ,Body3}} = Response3,

0 commit comments

Comments
 (0)