Skip to content

Commit

Permalink
add rabbitmq#732
Browse files Browse the repository at this point in the history
  • Loading branch information
@Anarh2404
Anarh2404 committed Feb 24, 2020
1 parent 99d5474 commit 2690724
Show file tree
Hide file tree
Showing 17 changed files with 644 additions and 545 deletions.
667 changes: 321 additions & 346 deletions projects/client/Apigen/src/apigen/Apigen.cs
View file

Large diffs are not rendered by default.

8 changes: 4 additions & 4 deletions projects/client/RabbitMQ.Client/src/client/impl/BasicPublishBatch.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -59,9 +59,9 @@ public void Add(string exchange, string routingKey, bool mandatory, IBasicProper
var bp = basicProperties == null ? model.CreateBasicProperties() : basicProperties;
var method = new BasicPublish
{
m_exchange = exchange,
m_routingKey = routingKey,
m_mandatory = mandatory
_exchange = exchange,
_routingKey = routingKey,
_mandatory = mandatory
};

commands.Add(new Command(method, (ContentHeaderBase)bp, body));
Expand All @@ -72,4 +72,4 @@ public void Publish()
model.SendCommands(commands);
}
}
}
}
18 changes: 6 additions & 12 deletions projects/client/RabbitMQ.Client/src/client/impl/CommandAssembler.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -81,11 +81,8 @@ public Command HandleFrame(InboundFrame f)
{
throw new UnexpectedFrameException(f);
}
BinaryBufferReader reader = f.GetReader();
m_method = m_protocol.DecodeMethodFrom(reader);
m_state = m_method.HasContent
? AssemblyState.ExpectingContentHeader
: AssemblyState.Complete;
m_method = m_protocol.DecodeMethodFrom(f.Payload.AsMemory(0, f.PayloadSize));
m_state = m_method.HasContent ? AssemblyState.ExpectingContentHeader : AssemblyState.Complete;
return CompletedCommand();
}
case AssemblyState.ExpectingContentHeader:
Expand All @@ -94,9 +91,9 @@ public Command HandleFrame(InboundFrame f)
{
throw new UnexpectedFrameException(f);
}
BinaryBufferReader reader = f.GetReader();
m_header = m_protocol.DecodeContentHeaderFrom(reader);
ulong totalBodyBytes = m_header.ReadFrom(reader);
Memory<byte> memory = f.Payload.AsMemory(0, f.PayloadSize);
m_header = m_protocol.DecodeContentHeaderFrom(NetworkOrderDeserializer.ReadUInt16(memory));
ulong totalBodyBytes = m_header.ReadFrom(memory.Slice(2));
if (totalBodyBytes > MaxArrayOfBytesSize)
{
throw new UnexpectedFrameException(f);
Expand All @@ -114,10 +111,7 @@ public Command HandleFrame(InboundFrame f)
}
if (f.PayloadSize > m_remainingBodyBytes)
{
throw new MalformedFrameException
(string.Format("Overlong content body received - {0} bytes remaining, {1} bytes received",
m_remainingBodyBytes,
f.PayloadSize));
throw new MalformedFrameException($"Overlong content body received - {m_remainingBodyBytes} bytes remaining, {f.PayloadSize} bytes received");
}
f.PayloadSpan.CopyTo(UnwritedSpan);
m_remainingBodyBytes -= f.PayloadSize;
Expand Down
40 changes: 20 additions & 20 deletions projects/client/RabbitMQ.Client/src/client/impl/Connection.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ public class Connection : IConnection

private IDictionary<string, object> m_clientProperties;

private volatile ShutdownEventArgs m_closeReason = null;
private volatile ShutdownEventArgs _closeReason = null;
private volatile bool m_closed = false;

private EventHandler<ConnectionBlockedEventArgs> m_connectionBlocked;
Expand All @@ -85,7 +85,7 @@ public class Connection : IConnection
private Guid m_id = Guid.NewGuid();
private ModelBase m_model0;
private volatile bool m_running = true;
private MainSession m_session0;
private MainSession _session0;
private SessionManager m_sessionManager;

private IList<ShutdownReportEntry> m_shutdownReport = new SynchronizedList<ShutdownReportEntry>(new List<ShutdownReportEntry>());
Expand Down Expand Up @@ -132,8 +132,8 @@ public Connection(IConnectionFactory factory, bool insist, IFrameHandler frameHa
}

m_sessionManager = new SessionManager(this, 0);
m_session0 = new MainSession(this) { Handler = NotifyReceivedCloseOk };
m_model0 = (ModelBase)Protocol.CreateModel(m_session0);
_session0 = new MainSession(this) { Handler = NotifyReceivedCloseOk };
m_model0 = (ModelBase)Protocol.CreateModel(_session0);

StartMainLoop(factory.UseBackgroundThreadsForIO);
Open(insist);
Expand Down Expand Up @@ -202,15 +202,15 @@ public event EventHandler<ShutdownEventArgs> ConnectionShutdown
bool ok = false;
lock (m_eventLock)
{
if (m_closeReason == null)
if (_closeReason == null)
{
m_connectionShutdown += value;
ok = true;
}
}
if (!ok)
{
value(this, m_closeReason);
value(this, _closeReason);
}
}
remove
Expand Down Expand Up @@ -273,7 +273,7 @@ public IDictionary<string, object> ClientProperties

public ShutdownEventArgs CloseReason
{
get { return m_closeReason; }
get { return _closeReason; }
}

public AmqpTcpEndpoint Endpoint
Expand Down Expand Up @@ -386,19 +386,19 @@ public void Close(ShutdownEventArgs reason, bool abort, TimeSpan timeout)
{
if (!abort)
{
throw new AlreadyClosedException(m_closeReason);
throw new AlreadyClosedException(_closeReason);
}
}
else
{
OnShutdown();
m_session0.SetSessionClosing(false);
_session0.SetSessionClosing(false);

try
{
// Try to send connection.close
// Wait for CloseOk in the MainLoop
m_session0.Transmit(ConnectionCloseWrapper(reason.ReplyCode, reason.ReplyText));
_session0.Transmit(ConnectionCloseWrapper(reason.ReplyCode, reason.ReplyText));
}
catch (AlreadyClosedException ace)
{
Expand Down Expand Up @@ -521,7 +521,7 @@ public void FinishClose()
MaybeStopHeartbeatTimers();

m_frameHandler.Close();
m_model0.SetCloseReason(m_closeReason);
m_model0.SetCloseReason(_closeReason);
m_model0.FinishClose();
}

Expand Down Expand Up @@ -552,10 +552,10 @@ public bool HardProtocolExceptionHandler(HardProtocolException hpe)
if (SetCloseReason(hpe.ShutdownReason))
{
OnShutdown();
m_session0.SetSessionClosing(false);
_session0.SetSessionClosing(false);
try
{
m_session0.Transmit(ConnectionCloseWrapper(
_session0.Transmit(ConnectionCloseWrapper(
hpe.ShutdownReason.ReplyCode,
hpe.ShutdownReason.ReplyText));
return true;
Expand All @@ -580,13 +580,13 @@ public void InternalClose(ShutdownEventArgs reason)
{
if (m_closed)
{
throw new AlreadyClosedException(m_closeReason);
throw new AlreadyClosedException(_closeReason);
}
// We are quiescing, but still allow for server-close
}

OnShutdown();
m_session0.SetSessionClosing(true);
_session0.SetSessionClosing(true);
TerminateMainloop();
}

Expand Down Expand Up @@ -688,7 +688,7 @@ public async Task MainLoopIteration()
// quiescing situation, even though technically we
// should be ignoring everything except
// connection.close-ok.
m_session0.HandleFrame(frame);
_session0.HandleFrame(frame);
}
else
{
Expand All @@ -697,7 +697,7 @@ public async Task MainLoopIteration()
// frames for non-zero channels (and any inbound
// commands on channel zero that aren't
// Connection.CloseOk) must be discarded.
if (m_closeReason == null)
if (_closeReason == null)
{
// No close reason, not quiescing the
// connection. Handle the frame. (Of course, the
Expand Down Expand Up @@ -819,7 +819,7 @@ public void OnShutdown()
lock (m_eventLock)
{
handler = m_connectionShutdown;
reason = m_closeReason;
reason = _closeReason;
m_connectionShutdown = null;
}
if (handler != null)
Expand Down Expand Up @@ -929,9 +929,9 @@ public bool SetCloseReason(ShutdownEventArgs reason)
{
lock (m_eventLock)
{
if (m_closeReason == null)
if (_closeReason == null)
{
m_closeReason = reason;
_closeReason = reason;
return true;
}
else
Expand Down
10 changes: 5 additions & 5 deletions projects/client/RabbitMQ.Client/src/client/impl/ContentHeaderBase.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -66,12 +66,12 @@ public virtual object Clone()
///<summary>
/// Fill this instance from the given byte buffer stream.
///</summary>
public ulong ReadFrom(BinaryBufferReader reader)
public ulong ReadFrom(Memory<byte> memory)
{
reader.ReadUInt16(); // weight - not currently used
ulong bodySize = reader.ReadUInt64();
var headerReader = new ContentHeaderPropertyReader(reader);
ReadPropertiesFrom(ref headerReader);
// Skipping the first two bytes since they arent used (weight - not currently used)
ulong bodySize = NetworkOrderDeserializer.ReadUInt64(memory.Slice(2));
var reader = new ContentHeaderPropertyReader(memory.Slice(10));
ReadPropertiesFrom(ref reader);
return bodySize;
}

Expand Down
41 changes: 29 additions & 12 deletions projects/client/RabbitMQ.Client/src/client/impl/ContentHeaderPropertyReader.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -49,14 +49,16 @@ public struct ContentHeaderPropertyReader
private ushort m_bitCount;
private ushort m_flagWord;

public ContentHeaderPropertyReader(BinaryBufferReader reader)
public ContentHeaderPropertyReader(Memory<byte> memory)
{
BaseReader = reader;
_memory = memory;
m_flagWord = 1; // just the continuation bit
m_bitCount = 15; // the correct position to force a m_flagWord read
_memoryOffset = 0;
}

private BinaryBufferReader BaseReader;
private Memory<byte> _memory;
private int _memoryOffset;

public bool ContinuationBitSet
{
Expand All @@ -82,28 +84,35 @@ public void ReadFlagWord()
{
throw new MalformedFrameException("Attempted to read flag word when none advertised");
}
m_flagWord = BaseReader.ReadUInt16();
m_flagWord = NetworkOrderDeserializer.ReadUInt16(_memory.Slice(_memoryOffset));
_memoryOffset += 2;
m_bitCount = 0;
}

public uint ReadLong()
{
return WireFormatting.ReadLong(BaseReader);
uint result = NetworkOrderDeserializer.ReadUInt32(_memory.Slice(_memoryOffset));
_memoryOffset += 4;
return result;
}

public ulong ReadLonglong()
{
return WireFormatting.ReadLonglong(BaseReader);
ulong result = NetworkOrderDeserializer.ReadUInt64(_memory.Slice(_memoryOffset));
_memoryOffset += 8;
return result;
}

public byte[] ReadLongstr()
{
return WireFormatting.ReadLongstr(BaseReader);
byte[] result = WireFormatting.ReadLongstr(_memory.Slice(_memoryOffset));
_memoryOffset += 4 + result.Length;
return result;
}

public byte ReadOctet()
{
return WireFormatting.ReadOctet(BaseReader);
return _memory.Span[_memoryOffset++];
}

public bool ReadPresence()
Expand All @@ -121,23 +130,31 @@ public bool ReadPresence()

public ushort ReadShort()
{
return WireFormatting.ReadShort(BaseReader);
ushort result = NetworkOrderDeserializer.ReadUInt16(_memory.Slice(_memoryOffset));
_memoryOffset += 2;
return result;
}

public string ReadShortstr()
{
return WireFormatting.ReadShortstr(BaseReader);
string result = WireFormatting.ReadShortstr(_memory.Slice(_memoryOffset), out int bytesRead);
_memoryOffset += bytesRead;
return result;
}

/// <returns>A type of <seealso cref="System.Collections.Generic.IDictionary{TKey,TValue}"/>.</returns>
public IDictionary<string, object> ReadTable()
{
return WireFormatting.ReadTable(BaseReader);
IDictionary<string, object> result = WireFormatting.ReadTable(_memory.Slice(_memoryOffset), out int bytesRead);
_memoryOffset += bytesRead;
return result;
}

public AmqpTimestamp ReadTimestamp()
{
return WireFormatting.ReadTimestamp(BaseReader);
AmqpTimestamp result = WireFormatting.ReadTimestamp(_memory.Slice(_memoryOffset));
_memoryOffset += 8;
return result;
}
}
}
2 changes: 1 addition & 1 deletion projects/client/RabbitMQ.Client/src/client/impl/MainSession.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ public override void HandleFrame(InboundFrame frame)

if (!m_closeServerInitiated && (frame.IsMethod()))
{
MethodBase method = Connection.Protocol.DecodeMethodFrom(frame.GetReader());
MethodBase method = Connection.Protocol.DecodeMethodFrom(frame.Payload.AsMemory(0, frame.PayloadSize));
if ((method.ProtocolClassId == m_closeClassId)
&& (method.ProtocolMethodId == m_closeMethodId))
{
Expand Down
Loading

0 comments on commit 2690724

Please sign in to comment.