indexing description: "Provides Discovery Service" license: "MIT license (see ../../license.txt)" author: "Beat Strasser " date: "$Date$" revision: "$Revision$" class P2P_DISCOVERY_SERVICE inherit P2P_MODULE redefine init end THREAD rename execute as srdi_manager export {NONE} all end DT_SHARED_SYSTEM_CLOCK export {NONE} all end P2P_STRING_UTILS P2P_EXCEPTION_LOG create init feature {NONE} -- Initialization init (group: P2P_PEERGROUP; id: P2P_ID; advertisement: like implementation_advertisement) is -- Initialize module do Precursor (group, id, advertisement) cm := peer_group.cache_manager resolver := peer_group.resolver_service resolver_handler_name := module_id.out local_only := False -- maybe changed in the future (to be configured in platform configuration) -- as long as the endpoint router doesn't look automatically for peer/route advertisements, -- it doesn't make sense to keep a certain capacity limit and flush old advs, so no limits here -- for the moment. create peers_cache.make_without_capacity_limit create groups_cache.make_without_capacity_limit create generals_cache.make_without_capacity_limit create query_listeners.make_default create response_listeners.make query_id := 0 -- load persistent cache into LRU cache load_advertisements end feature -- Access local_only: BOOLEAN Type_peer: INTEGER is 0 Type_group: INTEGER is 1 Type_general: INTEGER is 2 Default_expiration_time: INTEGER_64 is 7200000 -- 2 hours resolver_handler_name: STRING feature -- Discovery local_peer_advertisement (a_peer_id: P2P_PEER_ID): P2P_PEER_ADVERTISEMENT is -- Find peer advertisement for `a_peer_id' require Status_ok: module_status = initializing or module_status = start_ok or module_status = suspending or module_status = suspended Id_valid: a_peer_id /= Void and a_peer_id.is_valid do Result := peers_cache.find_advertisement (a_peer_id.out_short) logger.debugging ("discovery_service: Queried for local peer advertisement, unique id: " + a_peer_id.out_short + ", found: " + (Result /= Void).out + ", group id: " + peer_group.id.out) ensure Result_valid: Result /= Void implies Result.unique_id.is_equal (a_peer_id.out_short) end local_peer_advertisements (element_name, element_value: STRING): DS_LIST [P2P_PEER_ADVERTISEMENT] is -- Find local peer advertisements require Status_ok: module_status = initializing or module_status = start_ok or module_status = suspending or module_status = suspended Name_set: element_name /= Void do Result := peers_cache.find_matches (element_name, element_value) logger.debugging ("discovery_service: Queried for local peer advertisements, key: " + element_name + ", value: " + to_string (element_value) + ", results: " + Result.count.out + ", group id: " + peer_group.id.out) ensure Result_set: Result /= Void end local_group_advertisement (a_group_id: P2P_PEERGROUP_ID): P2P_PEERGROUP_ADVERTISEMENT is -- Find peer group advertisement for `a_group_id' require Status_ok: module_status = initializing or module_status = start_ok or module_status = suspending or module_status = suspended Id_valid: a_group_id /= Void and a_group_id.is_valid do Result := groups_cache.find_advertisement (a_group_id.out_short) logger.debugging ("discovery_service: Queried for local peergroup advertisement, unique id: " + a_group_id.out_short + ", found: " + (Result /= Void).out + ", group id: " + peer_group.id.out) ensure Result_valid: Result /= Void implies Result.unique_id.is_equal (a_group_id.out_short) end local_group_advertisements (element_name, element_value: STRING): DS_LIST [P2P_PEERGROUP_ADVERTISEMENT] is -- Find local peer group advertisements require Status_ok: module_status = initializing or module_status = start_ok or module_status = suspending or module_status = suspended Name_set: element_name /= Void do Result := groups_cache.find_matches (element_name, element_value) logger.debugging ("discovery_service: Queried for local group advertisements, key: " + element_name + ", value: " + to_string (element_value) + ", results: " + Result.count.out + ", group id: " + peer_group.id.out) ensure Result_set: Result /= Void end local_general_advertisements (element_name, element_value: STRING): DS_LIST [P2P_ADVERTISEMENT] is -- Find local general advertisements (not peer/peergroup advertisements) require Status_ok: module_status = initializing or module_status = start_ok or module_status = suspending or module_status = suspended Name_set: element_name /= Void do Result := generals_cache.find_matches (element_name, element_value) logger.debugging ("discovery_service: Queried for local general advertisements, key: " + element_name + ", value: " + to_string (element_value) + ", results: " + Result.count.out + ", group id: " + peer_group.id.out) ensure Result_set: Result /= Void end local_general_advertisement (a_unique_id: STRING): P2P_ADVERTISEMENT is -- Find advertisement with `a_unique_id' require Status_ok: module_status = initializing or module_status = start_ok or module_status = suspending or module_status = suspended Id_valid: a_unique_id /= Void do Result := generals_cache.find_advertisement (a_unique_id) logger.debugging ("discovery_service: Queried for local general advertisement, unique id: " + a_unique_id + ", found: " + (Result /= Void).out + ", group id: " + peer_group.id.out) ensure Result_valid: Result /= Void implies Result.unique_id.is_equal (a_unique_id) end query_remote_advertisements (a_query: P2P_DISCOVERY_QUERY; a_dest_peer_id: P2P_PEER_ID; a_handler: PROCEDURE [ANY, TUPLE [P2P_DISCOVERY_RESPONSE]]) is -- Find remote peer advertisements and set queries corresponding resolver query id require Status_ok: module_status = start_ok or module_status = suspending or module_status = suspended Not_local_only: not local_only Query_valid: a_query /= Void and a_query.is_valid Dest_peer_valid: a_dest_peer_id /= Void implies a_dest_peer_id.is_valid local resolver_query: P2P_RESOLVER_QUERY do logger.debugging ("discovery_service: Querying for remote advertisements, type: " + a_query.type.out + ", key: " + to_string (a_query.key) + ", value: " + to_string (a_query.value) + ", destination peer: " + to_string (a_dest_peer_id) + ", group id: " + peer_group.id.out) if a_query.resolver_query_id <= 0 then -- goto next query id increment_query_id a_query.set_resolver_query_id (query_id) end -- create resolver query create resolver_query.make (peer_group.peer_id, resolver_handler_name, a_query.resolver_query_id, a_query.out) -- register listener for query id if a_handler /= Void then query_listeners.force (a_handler, a_query.resolver_query_id) end -- send resolver query if a_dest_peer_id /= Void then resolver.send_query (a_dest_peer_id, resolver_query) else resolver.propagate_query (resolver_query) end end process_query (a_query: P2P_RESOLVER_QUERY) is -- Process incoming discovery query require Status_ok: module_status = start_ok or module_status = suspending or module_status = suspended Query_valid: a_query /= Void and a_query.is_valid and a_query.handler_name.is_equal (resolver_handler_name) local discovery_query: P2P_DISCOVERY_QUERY discovery_response: P2P_DISCOVERY_RESPONSE resolver_response: P2P_RESOLVER_RESPONSE responses: DS_LIST [P2P_ADVERTISEMENT] do -- parse discovery query create discovery_query.parse_from_string (a_query.query) if discovery_query.is_valid then -- publish remote's peer advertisement if discovery_query.peer_advertisement /= Void then publish_advertisement_locally (discovery_query.peer_advertisement) end if discovery_query.threshold = 0 and discovery_query.type = type_peer then logger.info ("discovery_service: Incoming query processed, sending back our peer advertisement, originator: " + a_query.source_peer_id.out + ", group id: " + peer_group.id.out) -- just return our peer advertisement create discovery_response.make_with_peer_advertisement (peer_group.peer_advertisement) create resolver_response.make_from_query (a_query, peer_group.peer_id, discovery_response.out) resolver.send_response (a_query.source_peer_id, resolver_response) elseif discovery_query.threshold /= 0 then if discovery_query.key = Void and discovery_query.value = Void then -- looking for random set of advertisements? if discovery_query.type = type_peer then responses := peers_cache.find_all (discovery_query.threshold) elseif discovery_query.type = type_group then responses := groups_cache.find_all (discovery_query.threshold) else -- type = type_general responses := generals_cache.find_all (discovery_query.threshold) end else -- look for known matching advertisements if discovery_query.type = type_peer then responses := peers_cache.find_matches (discovery_query.key, discovery_query.value) elseif discovery_query.type = type_group then responses := groups_cache.find_matches (discovery_query.key, discovery_query.value) else -- type = type_general responses := generals_cache.find_matches (discovery_query.key, discovery_query.value) end end if responses.count > 0 then logger.info ("discovery_service: Incoming query processed, sending back matching advertisements, count: " + responses.count.out + ", type: " + discovery_query.type.out + ", key: " + to_string (discovery_query.key) + ", value: " + to_string (discovery_query.value) + ", group id: " + peer_group.id.out) create discovery_response.make_from_query (discovery_query, responses) create resolver_response.make_from_query (a_query, peer_group.peer_id, discovery_response.out) resolver.send_response (a_query.source_peer_id, resolver_response) else logger.info ("discovery_service: Incoming query processed, no matching advertisements, type: " + discovery_query.type.out + ", key: " + to_string (discovery_query.key) + ", value: " + to_string (discovery_query.value) + ", group id: " + peer_group.id.out) end -- repropagate query (resolver will check if we're a rendezvous) a_query.set_repropagate (True) end end end feature -- Publishing publish_advertisement_locally (adv: P2P_ADVERTISEMENT) is -- Publish local advertisement require Status_ok: module_status = initializing or module_status = start_ok or module_status = suspending or module_status = suspended Advertisement_set: adv /= Void and adv.is_valid local peer_adv: P2P_PEER_ADVERTISEMENT group_adv: P2P_PEERGROUP_ADVERTISEMENT route_adv: P2P_ROUTE_ADVERTISEMENT do -- peer advertisement? peer_adv ?= adv if peer_adv /= Void then peers_cache.put (peer_adv) -- update a possible single route advertisement for this peer route_adv ?= peer_adv.service_parameter (endpoint_mcid) if route_adv /= Void then route_adv.set_destination_peer_id (peer_adv.peer_id) if generals_cache.find_advertisement (route_adv.unique_id) /= Void then generals_cache.put (route_adv) end end logger.debugging ("discovery_service: Published peer advertisement locally, peer id: " + peer_adv.peer_id.out + ", group id: " + peer_group.id.out) else -- group advertisement? group_adv ?= adv if group_adv /= Void then groups_cache.put (group_adv) logger.debugging ("discovery_service: Published group advertisement locally, adv group id: " + group_adv.group_id.out + ", group id: " + peer_group.id.out) else -- general advertisement generals_cache.put (adv) logger.debugging ("discovery_service: Published general advertisement locally, key: " + to_string (adv.unique_id) + ", group id: " + peer_group.id.out) end end end publish_advertisement_remotely (an_adv: P2P_ADVERTISEMENT; a_dest_peer_id: P2P_PEER_ID) is -- Send advertisement to peer/group. Don't store it locally. require Status_ok: module_status = start_ok or module_status = suspending or module_status = suspended Not_local_only: not local_only Advertisement_valid: an_adv /= Void and an_adv.is_valid Peer_valid: a_dest_peer_id /= Void implies a_dest_peer_id.is_valid local peer_adv: P2P_PEER_ADVERTISEMENT group_adv: P2P_PEERGROUP_ADVERTISEMENT discovery_response: P2P_DISCOVERY_RESPONSE resolver_response: P2P_RESOLVER_RESPONSE responses_list: DS_ARRAYED_LIST [P2P_ADVERTISEMENT] do -- create discovery response peer_adv ?= an_adv if peer_adv /= Void then create discovery_response.make_with_peer_advertisement (peer_adv) else create responses_list.make (1) responses_list.put_first (an_adv) group_adv ?= an_adv if group_adv /= Void then create discovery_response.make_with_response (type_group, responses_list) else create discovery_response.make_with_response (type_general, responses_list) end end -- create/send resolver response message create resolver_response.make (peer_group.peer_id, resolver_handler_name, 0, discovery_response.out) if a_dest_peer_id = Void then resolver.propagate_response (resolver_response) else resolver.send_response (a_dest_peer_id, resolver_response) end logger.info ("discovery_service: Published advertisement remotely, key: " + to_string (an_adv.unique_id) + ", destination peer: " + to_string (a_dest_peer_id) + ", group id: " + peer_group.id.out) end publish_advertisements_remotely (a_response: P2P_DISCOVERY_RESPONSE; a_dest_peer_id: P2P_PEER_ID) is -- Send `a_response' to peer/group. Don't store it locally. require Status_ok: module_status = start_ok or module_status = suspending or module_status = suspended Not_local_only: not local_only Response_valid: a_response /= Void and a_response.is_valid Peer_valid: a_dest_peer_id /= Void implies a_dest_peer_id.is_valid local resolver_response: P2P_RESOLVER_RESPONSE do -- create/send resolver response message create resolver_response.make (peer_group.peer_id, resolver_handler_name, 0, a_response.out) if a_dest_peer_id = Void then resolver.propagate_response (resolver_response) else resolver.send_response (a_dest_peer_id, resolver_response) end logger.info ("discovery_service: Published advertisements remotely, responses: " + a_response.count.out + ", destination peer: " + to_string (a_dest_peer_id) + ", group id: " + peer_group.id.out) end process_response (a_response: P2P_RESOLVER_RESPONSE) is -- Process incoming discovery response require Status_ok: module_status = start_ok or module_status = suspending or module_status = suspended Response_valid: a_response /= Void and a_response.is_valid and a_response.handler_name.is_equal (resolver_handler_name) local discovery_response: P2P_DISCOVERY_RESPONSE adv_cursor: DS_LIST_CURSOR [P2P_ADVERTISEMENT] listener_cursor: DS_LINKED_LIST_CURSOR [PROCEDURE [ANY, TUPLE [P2P_DISCOVERY_RESPONSE]]] response_handled: BOOLEAN do -- parse discovery response create discovery_response.parse_from_string (a_response.response) if discovery_response.is_valid then logger.info ("discovery_service: Processing incoming discovery response, originator: " + to_string (a_response.response_peer_id) + ", responses: " + discovery_response.count.out + ", group id: " + peer_group.id.out) -- call query id listener if query_listeners.has (a_response.query_id) then call_handler (query_listeners.item (a_response.query_id), discovery_response) response_handled := True end -- call general listeners from listener_cursor := response_listeners.new_cursor listener_cursor.start until listener_cursor.after loop call_handler (listener_cursor.item, discovery_response) response_handled := True if not listener_cursor.after then listener_cursor.forth end end if not response_handled then logger.debugging ("discovery_service: Publishing incoming discovery responses (no handlers found), responses: " + discovery_response.count.out + ", group id: " + peer_group.id.out) -- publish advertisements locally, if response not handled if discovery_response.peer_advertisement /= Void then publish_advertisement_locally (discovery_response.peer_advertisement) end from adv_cursor := discovery_response.responses.new_cursor adv_cursor.start until adv_cursor.after loop publish_advertisement_locally (adv_cursor.item) adv_cursor.forth end end end end feature -- Removal flush_peer_advertisement (a_peer_id: P2P_PEER_ID) is -- Flush peer advertisement for `a_peer_id' require Status_ok: module_status = initializing or module_status = start_ok or module_status = suspending or module_status = suspended Peer_id_valid: a_peer_id /= Void and a_peer_id.is_valid do peers_cache.remove (a_peer_id.out_short) logger.debugging ("discovery_service: Flushed local peer advertisement, unique id: " + a_peer_id.out_short + ", group id: " + peer_group.id.out) end flush_peergroup_advertisement (a_group_id: P2P_PEERGROUP_ID) is -- Flush peer group advertisement for `a_group_id' require Status_ok: module_status = initializing or module_status = start_ok or module_status = suspending or module_status = suspended Group_id_valid: a_group_id /= Void and a_group_id.is_valid do groups_cache.remove (a_group_id.out_short) logger.debugging ("discovery_service: Flushed local peer group advertisement, unique id: " + a_group_id.out_short + ", group id: " + peer_group.id.out) end flush_general_advertisement (a_unique_id: STRING) is -- Flush general advertisement with `a_unique_id' require Status_ok: module_status = initializing or module_status = start_ok or module_status = suspending or module_status = suspended Unique_id_valid: a_unique_id /= Void do generals_cache.remove (a_unique_id) logger.debugging ("discovery_service: Flushed local general advertisement, unique id: " + a_unique_id + ", group id: " + peer_group.id.out) end feature -- Listeners remove_queryid_listener (a_handler_target: ANY) is -- Unregister corresponding query id listener for `a_handler_target' require Status_ok: module_status = start_ok or module_status = suspending or module_status = suspended Handler_valid: a_handler_target /= Void local cursor: DS_HASH_TABLE_CURSOR [PROCEDURE [ANY, TUPLE], INTEGER] do from cursor := query_listeners.new_cursor cursor.start until cursor.after loop if cursor.item.target = a_handler_target then query_listeners.remove (cursor.key) -- and move forth else cursor.forth end end end extend_response_listener (a_handler: PROCEDURE [ANY, TUPLE [P2P_DISCOVERY_RESPONSE]]) is -- Register `a_handler' for all valid incoming discovery responses require Status_ok: module_status = initializing or module_status = start_ok or module_status = suspending or module_status = suspended Handler_valid: a_handler /= Void do if not response_listeners.has (a_handler) then response_listeners.put_last (a_handler) end end remove_response_listener (a_handler_target: ANY) is -- Unregister corresponding response handler(s) for `a_handler_target' require Status_ok: module_status = initializing or module_status = start_ok or module_status = suspending or module_status = suspended Handler_target_valid: a_handler_target /= Void local cursor: DS_LINKED_LIST_CURSOR [PROCEDURE [ANY, TUPLE]] do from cursor := response_listeners.new_cursor cursor.start until cursor.after loop if cursor.item.target = a_handler_target then cursor.remove -- and move forth else cursor.forth end end end feature -- Basic operations start (args: ARRAY [STRING]) is -- Start module do if module_status = initializing then -- register resolver handler if not local_only then resolver.extend_handler (resolver_handler_name, agent process_query, agent process_response) end end if not local_only then srdi_last_push := Void -- make srdi push full index after each rdv reconnect peer_group.rendezvous_service.extend_rendezvous_event_handler ( agent (an_event: P2P_RENDEZVOUS_EVENT) do if an_event.type = an_event.type_reconnected_to_rendezvous then srdi_last_push := Void end end) -- start srdi manager thread launch end module_status := start_ok end suspend is -- Suspend module do if not local_only then -- stop SRDI manager thread module_status := suspending peer_group.rendezvous_service.prune_rendezvous_event_handler (Current) join end module_status := suspended end stop is -- Stop module do -- unregister resolver handler if not local_only then resolver.prune_handler (resolver_handler_name) end module_status := stop_ok end feature {NONE} -- SRDI Manager srdi_last_push: DT_DATE_TIME Srdi_manager_processing_interval: INTEGER is 120 -- 1 minute Srdi_manager_sleep_interval: INTEGER_64 is 500000000 -- 500ms srdi_manager is -- SRDI manager thread local failed: BOOLEAN step: INTEGER do if not failed and peer_group.when_fully_started then logger.debugging ("discovery_service: Starting SRDI manager thread, group id: " + peer_group.id.out) from step := 0 until module_status = suspending loop if peer_group.rendezvous_service.is_connected then if srdi_last_push = Void or step = 0 then -- send new advs to our rdv(s), if we're connected push_srdi (type_peer, srdi_last_push = Void) push_srdi (type_group, srdi_last_push = Void) push_srdi (type_general, srdi_last_push = Void) srdi_last_push := utc_system_clock.date_time_now step := 0 end else srdi_last_push := Void end step := (step + 1) \\ srdi_manager_processing_interval sleep (srdi_manager_sleep_interval) end end logger.debugging ("discovery_service: Stopped discovery SRDI manager, group id: " + peer_group.id.out) rescue failed := True log_exceptions retry end push_srdi (a_type: INTEGER; full: BOOLEAN) is -- Send SRD Index of `a_type' to rendezvous (full or delta since last push) require Type_valid: a_type = type_peer or a_type = type_group or a_type = type_general local entries: DS_LIST [TUPLE [STRING, INTEGER_64, STRING]] payload: P2P_GENERIC_SRDI resolver_msg: P2P_RESOLVER_SRDI pkey: STRING changed_since: DT_DATE_TIME do if not full then -- time of last SRDI push changed_since := srdi_last_push end -- get entries if a_type = type_peer then entries := peers_cache.srdi_entries (changed_since) pkey := "Peers" elseif a_type = type_group then entries := groups_cache.srdi_entries (changed_since) pkey := "Groups" elseif a_type = type_general then entries := generals_cache.srdi_entries (changed_since) pkey := "Adv" end if entries.count > 0 then -- send entries logger.info ("discovery_service: Pushing SRDI entries to rendezvous, primary key: " + pkey + ", full index: " + to_string (changed_since = Void) + ", entries: " + entries.count.out + ", group id: " + peer_group.id.out) create payload.make (peer_group.peer_id, pkey, 1, entries) create resolver_msg.make (resolver_handler_name, payload.out) resolver.propagate_srdi (resolver_msg) end end feature {NONE} -- Implementation resolver: P2P_RESOLVER_SERVICE query_id: INTEGER query_listeners: DS_HASH_TABLE [PROCEDURE [ANY, TUPLE [P2P_DISCOVERY_RESPONSE]], INTEGER] response_listeners: DS_LINKED_LIST [PROCEDURE [ANY, TUPLE [P2P_DISCOVERY_RESPONSE]]] peers_cache: P2P_ADVERTISEMENTS_LRUCACHE [P2P_PEER_ADVERTISEMENT] groups_cache: P2P_ADVERTISEMENTS_LRUCACHE [P2P_PEERGROUP_ADVERTISEMENT] generals_cache: P2P_ADVERTISEMENTS_LRUCACHE [P2P_ADVERTISEMENT] Peers_cache_capacity: INTEGER is 50 Groups_cache_capacity: INTEGER is 10 Generals_cache_capacity: INTEGER is 100 Threshold_max: INTEGER is -1 check_dependencies (a_parent: P2P_PEERGROUP): BOOLEAN is -- Are all needed dependencies met? do Result := a_parent.cache_manager /= Void and a_parent.resolver_service /= Void end increment_query_id is -- Go to next `query_id' do query_id := (query_id + 1) \\ {INTEGER_32}.max_value end call_handler (a_handler: PROCEDURE [ANY, TUPLE [P2P_DISCOVERY_RESPONSE]]; a_response: P2P_DISCOVERY_RESPONSE) is -- Call handler require Handler_valid: a_handler /= Void Response_valid: a_response /= Void and a_response.is_valid local failed: BOOLEAN do if not failed then a_handler.call ([a_response]) end rescue -- catch all handler exceptions failed := True log_exceptions retry end feature {NONE} -- Persistent cache cm: P2P_CACHE_MANAGER load_advertisements is -- Load advertisements from persistent disk cache. All advertisements from persistent cache -- won't ever expire. local groups: DS_LIST [P2P_PEERGROUP_ID] group_id: P2P_PEERGROUP_UUID group: P2P_PEERGROUP_ADVERTISEMENT peers: DS_LIST [P2P_PEER_ID] peer: P2P_PEER_ADVERTISEMENT generals: DS_LIST [STRING] general: P2P_ADVERTISEMENT do logger.debugging ("discovery_service: Loading advertisements from disk, group id: " + peer_group.id.out) groups := cm.peergroup_advertisement_ids from groups.start until groups.after loop group_id ?= groups.item_for_iteration if group_id /= Void then -- get group advertisement group := cm.peergroup_advertisement (group_id) if group /= Void then if not group_id.is_worldgroup_id and peer_group.id.is_equal (group_id.parent_peer_group_id) then -- add child group of our group groups_cache.load (group) end -- only add advertisements for this group if this is our peer group if peer_group.id.is_equal (group_id) then if peer_group.parent_group.id.is_worldgroup_id then -- (private) NPG is published in its own group (because there is no real parent group) groups_cache.load (group) end -- get group's peer advertisements peers := cm.peer_advertisement_ids (group_id) from peers.start until peers.after or peers_cache.is_full loop peer := cm.peer_advertisement (group_id, peers.item_for_iteration) if peer /= Void then peers_cache.load (peer) end peers.forth end -- get groups' other advertisements generals := cm.general_advertisement_ids (group_id) from generals.start until generals.after or generals_cache.is_full loop general := cm.general_advertisement (group_id, generals.item_for_iteration) if general /= Void then generals_cache.load (general) end generals.forth end end end end groups.forth end logger.info ("discovery_service: Loaded advertisements from disk, peer advs: " + peers_cache.count.out + ", group advs: " + groups_cache.count.out + ", general advs: " + generals_cache.count.out + ", group id: " + peer_group.id.out) end save_advertisements is -- Save advertisements to persistent disk cache. Expiration is not handled (this means -- saved advertisements won't ever expire when the platform is restarted). local peers_cursor: DS_LIST_CURSOR [P2P_PEER_ADVERTISEMENT] groups_cursor: DS_LIST_CURSOR [P2P_PEERGROUP_ADVERTISEMENT] generals_cursor: DS_LIST_CURSOR [P2P_ADVERTISEMENT] do -- Save peers advertisements cm.flush_peer_advertisements (peer_group.id) from peers_cursor := peers_cache.find_all (threshold_max).new_cursor peers_cursor.start until peers_cursor.after loop cm.store_peer_advertisement (peers_cursor.item) peers_cursor.forth end -- save peer group advertisements -- don't flush as we don't have all group advertisements in memory from groups_cursor := groups_cache.find_all (threshold_max).new_cursor groups_cursor.start until groups_cursor.after loop cm.store_peergroup_advertisement (groups_cursor.item) groups_cursor.forth end -- save general advertisements cm.flush_general_advertisements (peer_group.id) from generals_cursor := generals_cache.find_all (threshold_max).new_cursor generals_cursor.start until generals_cursor.after loop cm.store_general_advertisement (peer_group.id, generals_cursor.item) generals_cursor.forth end logger.info ("discovery_service: Saved advertisements to disk, group id: " + peer_group.id.out) end end