fixed build and added implementations for new read functions

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -17,6 +17,12 @@ namespace MQTTnet.AspNetCore
         {
             PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
             Connection = connection ?? throw new ArgumentNullException(nameof(connection));
+
+            if (Connection.Transport != null)
+            {
+                _input = Connection.Transport.Input;
+                _output = Connection.Transport.Output;
+            }
         }
         private PipeReader _input;
         private PipeWriter _output;
@@ -117,8 +123,9 @@ namespace MQTTnet.AspNetCore
 
         public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout, CancellationToken cancellationToken)
         {
-            var buffer = PacketFormatterAdapter.Encode(packet).AsMemory();
-            var output = Connection.Transport.Output;
+            var buffer = PacketFormatterAdapter.Encode(packet);
+            var msg = buffer.AsMemory();
+            var output = _output;
 
             await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
             try

Source/MQTTnet.AspnetCore/SpanBasedMqttPacketBodyReader.cs

@@ -1,4 +1,5 @@
-using MQTTnet.Formatter;
+using MQTTnet.Exceptions;
+using MQTTnet.Formatter;
 using System;
 using System.Buffers.Binary;
 using System.Text;
@@ -32,13 +33,6 @@ namespace MQTTnet.AspNetCore
             return _buffer.Slice(_offset).ToArray();
         }
 
-        public ushort ReadUInt16()
-        {
-            var result = BinaryPrimitives.ReadUInt16BigEndian(_buffer.Span.Slice(_offset));
-            _offset += 2;
-            return result;
-        }
-
         public byte[] ReadWithLengthPrefix()
         {
             return ReadSegmentWithLengthPrefix().ToArray();
@@ -66,27 +60,60 @@ namespace MQTTnet.AspNetCore
 
         public ushort ReadTwoByteInteger()
         {
-            throw new NotImplementedException();
+            var result = BinaryPrimitives.ReadUInt16BigEndian(_buffer.Span.Slice(_offset));
+            _offset += 2;
+            return result;
         }
 
         public uint ReadFourByteInteger()
         {
-            throw new NotImplementedException();
+            var result = BinaryPrimitives.ReadUInt32BigEndian(_buffer.Span.Slice(_offset));
+            _offset += 4;
+            return result;
         }
 
         public uint ReadVariableLengthInteger()
         {
-            throw new NotImplementedException();
+            var multiplier = 1;
+            var value = 0U;
+            byte encodedByte;
+
+            do
+            {
+                encodedByte = ReadByte();
+                value += (uint)((encodedByte & 127) * multiplier);
+
+                if (multiplier > 2097152)
+                {
+                    throw new MqttProtocolViolationException("Variable length integer is invalid.");
+                }
+
+                multiplier *= 128;
+            } while ((encodedByte & 128) != 0);
+
+            return value;
         }
 
         public bool ReadBoolean()
         {
-            throw new NotImplementedException();
+            var buffer = ReadByte();
+
+            if (buffer == 0)
+            {
+                return false;
+            }
+
+            if (buffer == 1)
+            {
+                return true;
+            }
+
+            throw new MqttProtocolViolationException("Boolean values can be 0 or 1 only.");
         }
 
         public void Seek(ulong position)
         {
-            throw new NotImplementedException();
+            _offset = (int)position;
         }
     }
 }