@@ -904,6 +904,14 @@ def handle_subscriptions_listen(body)
904904 return invalid_request_response ( "Invalid Request: subscriptions/listen requires an id" )
905905 end
906906
907+ # The id is echoed in every message of the stream, so one that cannot be written back as JSON
908+ # (a String holding bytes that are not valid UTF-8 parses, but does not generate) is refused here,
909+ # while the refusal can still answer with a null id, rather than surfacing as a failed write after
910+ # the stream has been registered.
911+ unless json_encodable? ( request_id )
912+ return invalid_request_response ( "Invalid Request: subscriptions/listen id must be valid UTF-8" )
913+ end
914+
907915 begin
908916 if RequestEnvelope . modern? ( params )
909917 RequestEnvelope . parse! ( params , request : params )
@@ -946,7 +954,43 @@ def handle_subscriptions_listen(body)
946954 return too_many_listen_subscriptions_response ( request_id )
947955 end
948956
949- [ 200 , SSE_HEADERS . dup , listen_sse_body ( request_id , honored_filter ( filter ) ) ]
957+ honored = honored_filter ( filter )
958+
959+ # The acknowledgement echoes the honored filter, whose resource URIs are client-supplied strings,
960+ # so it is encoded before the stream is committed to: a URI that cannot be written as JSON is
961+ # refused with a 400 instead of failing the first write after the stream was registered.
962+ unless ( acknowledgement = listen_acknowledgement_json ( request_id , honored ) )
963+ return json_rpc_error_response (
964+ status : 400 ,
965+ code : JsonRpcHandler ::ErrorCode ::INVALID_PARAMS ,
966+ message : "Invalid params: subscriptions/listen `notifications` must be valid UTF-8" ,
967+ id : request_id ,
968+ )
969+ end
970+
971+ [ 200 , SSE_HEADERS . dup , listen_sse_body ( request_id , honored , acknowledgement ) ]
972+ end
973+
974+ # The `notifications/subscriptions/acknowledged` frame for a listen stream as JSON, or `nil` when
975+ # the request id or the honored filter holds a string JSON cannot encode.
976+ def listen_acknowledgement_json ( request_id , honored )
977+ {
978+ jsonrpc : "2.0" ,
979+ method : Methods ::NOTIFICATIONS_SUBSCRIPTIONS_ACKNOWLEDGED ,
980+ params : {
981+ notifications : honored ,
982+ _meta : { RequestEnvelope ::SUBSCRIPTION_ID_META_KEY . to_sym => request_id } ,
983+ } ,
984+ } . to_json
985+ rescue JSON ::GeneratorError
986+ nil
987+ end
988+
989+ def json_encodable? ( value )
990+ value . to_json
991+ true
992+ rescue JSON ::GeneratorError
993+ false
950994 end
951995
952996 def listen_subscriptions_full?
@@ -1005,7 +1049,7 @@ def first
10051049 # The entry is keyed by an identifier minted here, not by the request id: that id is unique only among
10061050 # the requesting client's own in-flight requests, and two clients that pick the same one must each get
10071051 # their stream, stamped with the id they sent.
1008- def listen_sse_body ( request_id , honored )
1052+ def listen_sse_body ( request_id , honored , acknowledgement )
10091053 ListenStreamBody . new do |stream |
10101054 subscription_key = SecureRandom . uuid
10111055 subscription = nil
@@ -1029,15 +1073,6 @@ def listen_sse_body(request_id, honored)
10291073 if subscription . nil?
10301074 close_stream_safely ( stream )
10311075 else
1032- acknowledgement = {
1033- jsonrpc : "2.0" ,
1034- method : Methods ::NOTIFICATIONS_SUBSCRIPTIONS_ACKNOWLEDGED ,
1035- params : {
1036- notifications : honored ,
1037- _meta : { RequestEnvelope ::SUBSCRIPTION_ID_META_KEY . to_sym => request_id } ,
1038- } ,
1039- }
1040-
10411076 begin
10421077 acknowledged = subscription [ :write_mutex ] . synchronize do
10431078 next false if subscription [ :closed ]
0 commit comments