#if (!UNITY_WEBGL || UNITY_EDITOR) && !BESTHTTP_DISABLE_ALTERNATE_SSL && !BESTHTTP_DISABLE_HTTP2 && !BESTHTTP_DISABLE_WEBSOCKET using System; using System.Collections.Generic; using BestHTTP.Extensions; using BestHTTP.PlatformSupport.Memory; using BestHTTP.WebSocket; namespace BestHTTP.Connections.HTTP2 { public sealed class HTTP2WebSocketStream : HTTP2Stream { public override bool HasFrameToSend { get { // Don't let the connection sleep until return this.outgoing.Count > 0 || // we already booked at least one frame in advance (this.State == HTTP2StreamStates.Open && this.remoteWindow > 0 && this.lastReadCount > 0 && (this.overHTTP2.frames.Count > 0 || this.chunkQueue.Count > 0)); // we are in the middle of sending request data } } public override TimeSpan NextInteraction => this.overHTTP2.GetNextInteraction(); private OverHTTP2 overHTTP2; // local list of websocket header-data pairs private List> chunkQueue = new List>(); public HTTP2WebSocketStream(uint id, HTTP2Handler parentHandler, HTTP2SettingsManager registry, HPACKEncoder hpackEncoder) : base(id, parentHandler, registry, hpackEncoder) { } public override void Assign(HTTPRequest request) { base.Assign(request); this.overHTTP2 = request.Tag as OverHTTP2; this.overHTTP2.SetHTTP2Handler(this.parent); } protected override void ProcessIncomingDATAFrame(ref HTTP2FrameHeaderAndPayload frame, ref uint windowUpdate) { try { if (this.State != HTTP2StreamStates.HalfClosedLocal && this.State != HTTP2StreamStates.Open) { // ERROR! return; } this.downloaded += frame.PayloadLength; this.overHTTP2.OnReadThread(frame.Payload.AsBuffer((int)frame.PayloadOffset, (int)frame.PayloadLength)); // frame's buffer will be released later frame.DontUseMemPool = true; // Track received data, and if necessary(local window getting too low), send a window update frame if (this.localWindow < frame.PayloadLength) { HTTPManager.Logger.Error(nameof(HTTP2WebSocketStream), string.Format("[{0}] Frame's PayloadLength ({1:N0}) is larger then local window ({2:N0}). Frame: {3}", this.Id, frame.PayloadLength, this.localWindow, frame), this.Context, this.AssignedRequest.Context, this.parent.Context); } else this.localWindow -= frame.PayloadLength; if ((frame.Flags & (byte)HTTP2DataFlags.END_STREAM) != 0) this.isEndSTRReceived = true; if (this.isEndSTRReceived) { HTTPManager.Logger.Information(nameof(HTTP2WebSocketStream), string.Format("[{0}] All data arrived, data length: {1:N0}", this.Id, this.downloaded), this.Context, this.AssignedRequest.Context, this.parent.Context); // create a short living thread to process the downloaded data: PlatformSupport.Threading.ThreadedRunner.RunShortLiving(FinishRequest, this, this.dataView); this.dataView = null; if (this.State == HTTP2StreamStates.HalfClosedLocal) this.State = HTTP2StreamStates.Closed; else this.State = HTTP2StreamStates.HalfClosedRemote; } if (this.isEndSTRReceived || this.localWindow <= this.windowUpdateThreshold) windowUpdate += this.settings.MySettings[HTTP2Settings.INITIAL_WINDOW_SIZE] - this.localWindow - windowUpdate; } catch (Exception ex) { HTTPManager.Logger.Exception(nameof(HTTP2WebSocketStream), nameof(ProcessIncomingDATAFrame), ex, this.parent.Context); } } protected override void ProcessOpenState(List outgoingFrames) { try { // remote Window can be negative! See https://httpwg.org/specs/rfc7540.html#InitialWindowSize if (this.remoteWindow <= 0) { HTTPManager.Logger.Information(nameof(HTTP2WebSocketStream), string.Format("[{0}] Skipping data sending as remote Window is {1}!", this.Id, this.remoteWindow), this.Context, this.AssignedRequest.Context, this.parent.Context); return; } this.overHTTP2.PreReadCallback(); Int64 maxFragmentSize = Math.Min(BestHTTP.WebSocket.WebSocket.MaxFragmentSize, this.settings.RemoteSettings[HTTP2Settings.MAX_FRAME_SIZE]); Int64 maxFrameSize = Math.Min(maxFragmentSize, this.remoteWindow); if (chunkQueue.Count == 0) { if (this.overHTTP2.frames.TryDequeue(out var frame)) { this.overHTTP2._bufferedAmount -= (int)frame.Data.Count; frame.WriteTo((header, data) => chunkQueue.Add(new KeyValuePair(header, data)), (uint)maxFragmentSize, false, this.Context); } } while (this.remoteWindow >= 6 && chunkQueue.Count > 0) { var kvp = chunkQueue[0]; BufferSegment header = kvp.Key; BufferSegment data = kvp.Value; int minBytes = header.Count; int maxBytes = minBytes + data.Count; // remote window is less than the minimum we have to send, or // the frame has data but we have space only to send the websocket header if (this.remoteWindow < minBytes || (maxBytes > minBytes && this.remoteWindow == minBytes)) return; HTTP2FrameHeaderAndPayload headerFrame = new HTTP2FrameHeaderAndPayload(); headerFrame.Type = HTTP2FrameTypes.DATA; headerFrame.StreamId = this.Id; headerFrame.PayloadOffset = (uint)header.Offset; headerFrame.PayloadLength = (uint)header.Count; headerFrame.Payload = header.Data; headerFrame.DontUseMemPool = false; if (data.Count > 0) { HTTP2FrameHeaderAndPayload dataFrame = new HTTP2FrameHeaderAndPayload(); dataFrame.Type = HTTP2FrameTypes.DATA; dataFrame.StreamId = this.Id; var buff = data.Slice(data.Offset, (int)Math.Min(data.Count, maxFrameSize)); dataFrame.PayloadOffset = (uint)buff.Offset; dataFrame.PayloadLength = (uint)buff.Count; dataFrame.Payload = buff.Data; data = data.Slice(buff.Offset + buff.Count); if (data.Count == 0) chunkQueue.RemoveAt(0); else chunkQueue[0] = new KeyValuePair(header, data); // release the buffer only with the final frame and with the final frame's last data chunk bool isLast = (header.Data[header.Offset] & 0x7F) != 0 && chunkQueue.Count == 0; headerFrame.DontUseMemPool = dataFrame.DontUseMemPool = !isLast; this.outgoing.Enqueue(headerFrame); this.outgoing.Enqueue(dataFrame); } else { this.outgoing.Enqueue(headerFrame); chunkQueue.RemoveAt(0); } } } catch (Exception ex) { HTTPManager.Logger.Exception(nameof(HTTP2WebSocketStream), nameof(ProcessOpenState), ex, this.parent.Context); } } } } #endif