| … | … | @@ -10,23 +10,33 @@ signal peers(peers) |
| 10 | 10 | |
| 11 | 11 | export var url = "hyper://blog.mauve.moe/" |
| 12 | 12 | export var extension_name = "hyper-gossip-v1" |
| 13 | export var pool_size = 8 |
| 14 | export var queue_warning_size = 8 |
| 13 | 15 | |
| 14 | 16 | var peerListRequest = HyperRequest.new() |
| 15 | 17 | var eventSource = HyperEventSource.new() |
| 16 | | var broadcastRequest = HyperRequest.new() |
| 17 | 18 | |
| 18 | | var seen_messages = LRU.new() |
| 19 | # Pool of HyperRequest objects to send multiple broadcasts at once |
| 20 | var requestPool = [] |
| 21 | # Queue of requests that should be sent out |
| 19 | 22 | var requestQueue = [] |
| 20 | | var isRequesting = false |
| 23 | # Track whether we have warned about sending more requests than can be handled |
| 24 | var hasWarnedCapacity = false |
| 25 | var seen_messages = LRU.new() |
| 21 | 26 | |
| 22 | 27 | func _ready(): |
| 23 | 28 | randomize() |
| 24 | 29 | |
| 25 | 30 | add_child(peerListRequest) |
| 26 | 31 | peerListRequest.connect("request_completed", self, "_on_peer_list") |
| 32 | peerListRequest.use_threads = true |
| 27 | 33 | |
| 28 | | add_child(broadcastRequest) |
| 29 | | broadcastRequest.connect("request_completed", self, "_on_broadcast_completed") |
| 34 | for i in range(pool_size): |
| 35 | var broadcastRequest = HyperRequest.new() |
| 36 | broadcastRequest.use_threads = true |
| 37 | add_child(broadcastRequest) |
| 38 | broadcastRequest.connect("request_completed", self, "_on_broadcast_completed") |
| 39 | requestPool.push_back(broadcastRequest) |
| 30 | 40 | |
| 31 | 41 | add_child(eventSource) |
| 32 | 42 | eventSource.connect("event", self, "_on_event") |
| … | … | @@ -40,19 +50,28 @@ func _on_events_started(_statusCode, _responseHeaders): |
| 40 | 50 | emit_signal("listening", extension_name) |
| 41 | 51 | |
| 42 | 52 | func _enqueue_broadcast_request(reqURL, body): |
| 43 | | if isRequesting: |
| 53 | var broadcastRequest = _get_next_request() |
| 54 | if broadcastRequest == null: |
| 55 | print('Queueing') |
| 44 | 56 | requestQueue.push_back({"reqURL": reqURL, "body": body}) |
| 45 | | else: _perform_broadcast_request(reqURL, body) |
| 46 | | |
| 47 | | func _perform_broadcast_request(reqURL, body): |
| 48 | | isRequesting = true |
| 49 | | broadcastRequest.request(reqURL, [], false, HTTPClient.METHOD_POST, body) |
| 57 | if requestQueue.size() > queue_warning_size && !hasWarnedCapacity: |
| 58 | hasWarnedCapacity = true |
| 59 | print_debug("Warning: You have more gossip events than your queue_warning_size permits, either slow down the rate or increase the pool_size to send more"); |
| 60 | return |
| 61 | |
| 62 | broadcastRequest.request(reqURL, [], false, HTTPClient.METHOD_POST, body) |
| 63 | |
| 64 | func _get_next_request(): |
| 65 | for request in requestPool: |
| 66 | var status = request.get_http_client_status() |
| 67 | if status == HTTPClient.STATUS_DISCONNECTED: |
| 68 | return request |
| 69 | return null |
| 50 | 70 | |
| 51 | 71 | func _on_broadcast_completed(_result, _response_code, _headers, _body): |
| 52 | | isRequesting = false |
| 53 | 72 | if requestQueue.size() == 0: return |
| 54 | 73 | var next = requestQueue.pop_front() |
| 55 | | _perform_broadcast_request(next.reqURL, next.body) |
| 74 | _enqueue_broadcast_request(next.reqURL, next.body) |
| 56 | 75 | |
| 57 | 76 | func rebroadcast(event): |
| 58 | 77 | var reqURL = _get_extension_url() |
| … | … | @@ -75,7 +94,6 @@ func load_peers(): |
| 75 | 94 | |
| 76 | 95 | func _generateEvent(type, data): |
| 77 | 96 | var id = _make_id() |
| 78 | | print("generateEvent ID : " + String(id) ) |
| 79 | 97 | seen_messages.track(id) |
| 80 | 98 | return { |
| 81 | 99 | "id": id, |