Skip to content

Reworked network communication #30

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ChiefDesigner wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Reworked network communication #30

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
233 changes: 136 additions & 97 deletions Shem/BaseController.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,11 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Shem.AsyncEvents;
using Shem.Commands;
using Shem.Replies;
using Shem.Sockets;
using Shem.Utils;
using Shem.Exceptions;

namespace Shem
{
Expand All @@ -15,38 +14,46 @@ namespace Shem
/// If you wanna do things in the right way use the 'TorController'
/// class instead.
/// </summary>
public abstract class BaseController
public abstract class BaseController : IDisposable
{
/// <summary>
/// Contains type of reply which is currently received.
/// </summary>
private bool _receiveData;

protected ControlSocket controlSocket;

protected int sleep = 10;

protected uint responseTimeout = 1000;
/// <summary>
/// Reply which is currently constrocted from received data.
/// </summary>
private Reply _currentReply;

protected Task asyncEventsListener;
/// <summary>
/// Socket which receives data from TOR.
/// </summary>
private ControlSocket _controlSocket;

protected bool asyncEventsListenerStop = false;
/// <summary>
/// FIFO buffer for received regular replies.
/// </summary>
private Queue<Reply> _replyBuffer;

protected bool canListen = false;
/// <summary>
/// Eventhandle which signals the receiving of a regular reply.
/// </summary>
private EventWaitHandle _replyReceivedHdl;

/// <summary>
/// The time the library should wait for a reply.
/// The time (in milliseconds) the library should wait for a reply.
/// </summary>
protected uint ResponseTimeout
{
get { return responseTimeout; }
set { responseTimeout = value; }
}
protected uint ResponseTimeout { get; set; }

/// <summary>
/// Is the Controller connected to the server.
/// </summary>
protected bool Connected
{
get { return controlSocket != null ? controlSocket.Connected : false; } // Null reference exception sucks balls.
get { return _controlSocket != null ? _controlSocket.Connected : false; } // Null reference exception sucks balls.
}


/// <summary>
/// Construct a new TorController, used to control TOR
/// </summary>
Expand All @@ -55,119 +62,139 @@ protected bool Connected
/// <param name="connect">If the controller should connect just after the initialization</param>
protected BaseController(string address = "127.0.0.1", uint port = 9051, bool connect = true)
{
controlSocket = new ControlSocket(address, port, connect);
_receiveData = false;
ResponseTimeout = 1000;

_controlSocket = new ControlSocket(address, port, connect);
_controlSocket.OnLineReceived += ControlSocket_OnLineReceived;

canListen = true;
_replyReceivedHdl = new EventWaitHandle(false, EventResetMode.AutoReset);

asyncEventsListener = Task.Run(() => { ListenForAsyncEvents(); });
_replyBuffer = new Queue<Reply>();
}

private async void ListenForAsyncEvents()
/// <summary>
/// Handle a single line which was received by the ControlSocket.
/// </summary>
/// <param name="line">Received line (ending with CRLF)</param>
private void ControlSocket_OnLineReceived(string line)
{
while (!asyncEventsListenerStop)
if(_receiveData)
{
if (canListen)
if (line.Length > 0)
{
lock (controlSocket)
if (line[0] == '.')
{
if (controlSocket.ResponseAvailable)
if (line == ".\r\n")
{
// terminating sequence received
// RFC2821: If the line is composed of a single period, it is
// treated as the end of mail indicator.
_receiveData = false;
HandleFinishedReply();
return;
}
else
{
string rawReply = controlSocket.Receive();
List<TorEvent> asyncEvents = new List<TorEvent>();
List<Reply> replies = Reply.Parse(rawReply);
foreach (var r in replies)
{
try
{
asyncEvents.Add(TorEvent.Parse(r));
}
catch (Exception ex)
{
Logger.LogError(ex.Message);
}
}
AsyncEventDispatcher(asyncEvents);
// escaped line received
// RFC2821: If the first character is a period and there are
// other characters on the line, the first character
// is deleted
line = line.Substring(1);
}
}
}
await Task.Delay(250);

_currentReply.RawString += line;
}
else
{
_currentReply = new Reply(line);

if ((line[3] == ' ') || // EndReply
(line[3] == '-')) // MidReply
{
HandleFinishedReply();
}
else if (line[3] == '+') // DataReply
{
_receiveData = true;
}
else
{
throw new NullReplyCodeException(line, 3);
}
}
}

/// <summary>
/// Send a command and returns the reply as a raw string
/// Handle a completely received reply.
/// </summary>
/// <param name="command">The command to be sent</param>
/// <returns>Returns the raw string replied by the server</returns>
public virtual string SendRawCommand(TCCommand command)
private void HandleFinishedReply()
{
string rawReply = "";

canListen = false;

lock (controlSocket)
if (((int)_currentReply.Code >= 600) && ((int)_currentReply.Code < 700))
{
//Send the command
controlSocket.Send(command.Raw());

//Wait for response
int timeout = (int)ResponseTimeout / sleep;
int i = 1;
while (!controlSocket.ResponseAvailable && i < timeout)
// reply is async if first number in reply code is 6
AsyncEventDispatcher(TorEvent.Parse(_currentReply));
}
else
{
// all other replies are defined as synchronous
lock (_replyBuffer)
{
Thread.Sleep(sleep);
i++;
_replyBuffer.Enqueue(_currentReply);
}
//Read Response

rawReply = controlSocket.Receive();
_replyReceivedHdl.Set();
}

canListen = true;

return rawReply;
}

/// <summary>
///
/// Send a command and returns all replies
/// </summary>
/// <param name="command"></param>
/// <returns></returns>
/// <param name="command">The command to be sent</param>
/// <returns>List of all received replies</returns>
public virtual List<Reply> SendCommand(TCCommand command)
{
List<Reply> replies = Reply.Parse(SendRawCommand(command));
List<Reply> asyncEventsReply = new List<Reply>();
List<TorEvent> asyncEvents = new List<TorEvent>();
_controlSocket.Send(command.Raw());

List<Reply> replies = new List<Reply>();

foreach (var r in replies)
// receive replies until we get an end reply
for (;;)
{
if (r.Code == ReplyCodes.ASYNC_EVENT_NOTIFICATION)
lock (_replyBuffer)
{
asyncEventsReply.Add(r);
while (_replyBuffer.Count > 0)
{
Reply reply = _replyBuffer.Dequeue();
replies.Add(reply);
if (reply.RawString[3] == ' ')
{
// end reply received --> finished
return replies;
}
}
}
#if DEBUG
if (!_replyReceivedHdl.WaitOne())
#else
if(!_replyReceivedHdl.WaitOne((int)ResponseTimeout))
#endif
{
throw new TimeoutException("Timeout while receiving replies");
}
}

foreach (var e in asyncEventsReply)
{
replies.Remove(e);
asyncEvents.Add(TorEvent.Parse(e));
}

AsyncEventDispatcher(asyncEvents);

return replies;
}

protected abstract void AsyncEventDispatcher(TorEvent asyncEvent);

protected abstract void AsyncEventDispatcher(List<TorEvent> asyncEvents);

/// <summary>
/// Connect to the control port.
/// </summary>
protected void Connect()
{
controlSocket.Connect();
_controlSocket.Connect();
}

/// <summary>
Expand All @@ -178,20 +205,32 @@ public virtual void Close()
{
if (Connected)
{
SendCommand(new Quit());
_controlSocket.Close();
}
}

SendRawCommand(new Quit());
asyncEventsListenerStop = true;
if (asyncEventsListener.Exception == null)
{
asyncEventsListener.Wait();
}
controlSocket.Close();
public void Dispose(bool disposing)
{
Close();

if (disposing)
{
_controlSocket.Dispose();
_controlSocket = null;
_replyReceivedHdl.Dispose();
_replyReceivedHdl = null;
}
}

public void Dispose()
{
Dispose(true);
}

~BaseController()
{
Close();
Dispose(false);
}
}
}
Loading