Repository: dotnet/aspnetcore
File: src/SignalR/common/Shared/MessageBuffer.cs (C#)

Code before:
                    // TODO: consider a minimum time between sending these?

                    var sequenceId = _latestReceivedSequenceId;

                    _ackMessage.SequenceId = sequenceId;

                    await _writeLock.WaitAsync().ConfigureAwait(false);
                    try
                    {
                        _protocol.WriteMessage(_ackMessage, _writer);
                        await _writer.FlushAsync().ConfigureAwait(false);
                        _lastAckedId = sequenceId;
                    }
                    finally
                    {
                        _writeLock.Release();
                    }
                }
            }
        }
    }

    public ValueTask<FlushResult> WriteAsync(SerializedHubMessage hubMessage, CancellationToken cancellationToken)
    {
        // Default to HubInvocationMessage as that's the only type we use SerializedHubMessage for currently. Should harden this in the future.
        return WriteAsyncCore(hubMessage.Message?.GetType() ?? typeof(HubInvocationMessage), hubMessage.GetSerializedMessage(_protocol), cancellationToken);
    }

    public ValueTask<FlushResult> WriteAsync(HubMessage hubMessage, CancellationToken cancellationToken)
    {
        return WriteAsyncCore(hubMessage.GetType(), _protocol.GetMessageBytes(hubMessage), cancellationToken);
    }

    private async ValueTask<FlushResult> WriteAsyncCore(Type hubMessageType, ReadOnlyMemory<byte> messageBytes, CancellationToken cancellationToken)
    {
        // TODO: Add backpressure based on message count
        if (_bufferedByteCount > _bufferLimit)
        {
            // primitive backpressure if buffer is full
            while (await _waitForAck.Reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
            {
                if (_waitForAck.Reader.TryRead(out var count) && count < _bufferLimit)
                {
                    break;
                }
            }
        }

        // Avoid condition where last Ack position is the position we're currently writing into the buffer
        // If we wrote messages around the entire buffer before another Ack arrived we would end up reading the Ack position and writing over a buffered message
        _waitForAck.Reader.TryRead(out _);

Review comment (bug):
```suggestion
        return WriteAsyncCore(typeof(HubInvocationMessage), hubMessage.GetSerializedMessage(_protocol), cancellationToken);
```

If we're really confident with what the comment is saying, why not change it to the above? We could add a Debug.Assert that `hubMessage.Message is null || hubMessage.Message.GetType() == typeof(HubInvocationMessage)`.

Code after:
                    // TODO: consider a minimum time between sending these?

                    var sequenceId = _latestReceivedSequenceId;

                    _ackMessage.SequenceId = sequenceId;

                    await _writeLock.WaitAsync().ConfigureAwait(false);
                    try
                    {
                        _protocol.WriteMessage(_ackMessage, _writer);
                        await _writer.FlushAsync().ConfigureAwait(false);
                        _lastAckedId = sequenceId;
                    }
                    finally
                    {
                        _writeLock.Release();
                    }
                }
            }
        }
    }

    public ValueTask<FlushResult> WriteAsync(SerializedHubMessage hubMessage, CancellationToken cancellationToken)
    {
        // Default to HubInvocationMessage as that's the only type we use SerializedHubMessage for currently when Message is null. Should harden this in the future.
        return WriteAsyncCore(hubMessage.Message?.GetType() ?? typeof(HubInvocationMessage), hubMessage.GetSerializedMessage(_protocol), cancellationToken);
    }

    public ValueTask<FlushResult> WriteAsync(HubMessage hubMessage, CancellationToken cancellationToken)
    {
        return WriteAsyncCore(hubMessage.GetType(), _protocol.GetMessageBytes(hubMessage), cancellationToken);
    }

    private async ValueTask<FlushResult> WriteAsyncCore(Type hubMessageType, ReadOnlyMemory<byte> messageBytes, CancellationToken cancellationToken)
    {
        // TODO: Add backpressure based on message count
        if (_bufferedByteCount > _bufferLimit)
        {
            // primitive backpressure if buffer is full
            while (await _waitForAck.Reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
            {
                if (_waitForAck.Reader.TryRead(out var count) && count < _bufferLimit)
                {
                    break;
                }
            }
        }

        // Avoid condition where last Ack position is the position we're currently writing into the buffer
        // If we wrote messages around the entire buffer before another Ack arrived we would end up reading the Ack position and writing over a buffered message
        _waitForAck.Reader.TryRead(out _);