In the previous article, we discussed the concept of a "pump" in programming and its various applications, but without a practical demonstration. This article presents a UDP communication demo that implements text messaging and file transfer capabilities within a local network, similar to a LAN messenger. While the functionality isn't comprehensive, it sufficiently illustrates how "pumps" can be applied in code.
Let's recall the meaning of a "pump" from the previous article: first, it operates continuously; second, it transfers data from one location to another for others to use. The pump has a "pending" data container (buffer), and within its cycle, there can be a "preprocessing" area where data is handled before being passed along. Finally, the pump outputs data for consumption, completing the "data source" → "pump" → "consumer" process. The pump clearly serves as the driving force in this mechanism. We can see that the consumer's data usage process is contained within the pump cycle. If data processing takes too long and a pump cycle cannot return promptly, data will accumulate in the buffer, affecting the pump's efficiency. Two solutions exist:
-
- Avoid long-running operations during data processing;
-
- Instead of directly using data, first store the pump's output in another container (buffer), then process the buffer data in a separate thread.
Solution 1 is clearly superficial and doesn't solve real problems, while solution 2 is worth trying because it separates data processing from the pump cycle, prevetning it from directly affecting the pump's operation. In fact, when discussing the UDP communication structure in the previous article, we already implemented what solution 2 describes. I mentioned that we shouldn't place data processing and analysis directly within the "data receiving pump" because it would affect data reception efficiency. Later, I created a UDP communication structure diagram using three "pumps" (data receiving pump, data analysis pump, and data processing pump), as shown below:
As illustrated above, the "data analysis pump" uses data passed from the "data receiving pump," and the "data processing pump" uses data from the "data analysis pump." Neither directly consumes data; instead, they first store data in their respective buffers and then open new "pumps" (threads) to process the buffer data. This way, each stage doesn't interfere with the others' efficiency.
When implementing this UDP communication demo, I defined three "pump" classes, a data sender class, and a helper class:
- UDPSocket: Implements the data receiving pump and socket binding;
- DataAnalyse: Data analysis pump;
- DataDeal: Data processing pump;
- UDPSender: Data sending;
- Helper classes: Auxiliary functions.
Here, DataAnalyse is the consumer of data from the UDPSocket pump, DataDeal is the consumer of data from the DataAnalyse pump, and we (developers) are the consumers of data from the DataDeal pump.
It's important to note that by default, the "pump" in DataDeal won't pass any data, nor will the one in DataAnalyse. We need to implement our own DataDeal and DataAnalyse classes based on specific requirements, overriding the Deal and Analyse virtual methods. I've defined two classes: MyDataDeal and MyDataAnalyse. Here's the code:
/// <summary>
/// Custom data processing class that overrides the Deal method
/// Responsible for handling specific data types, delegating others to the base class
/// Can derive new classes from MyDataDeal and override Deal method further
/// </summary>
public class MyDataDeal : DataDeal
{
public event UserLoginEventHandler UserLogin;
public event LoginResponseEventHandler LoginResponse;
public event TextMessageReceivedEventHandler TextMessageReceived;
public event TextMessageResponseEventHandler TextMessageResponse;
public event UserLogoutEventHandler UserLogout;
public event FileTransferRequestEventHandler FileTransferRequest;
public event FileTransferResponseEventHandler FileTransferResponse;
public event FileDataReceivedEventHandler FileDataReceived;
public event FileTransferCompleteEventHandler FileTransferComplete;
public event FileTransferStoppedEventHandler FileTransferStopped;
public event FileDataAcknowledgmentEventHandler FileDataAcknowledgment;
public MyDataDeal(DataAnalyse dataAnalyse)
: base(dataAnalyse)
{
}
protected override void ProcessData(Data data)
{
switch (data.MessageType)
{
case MessageType.Login:
{
if (UserLogin != null)
{
UserLogin(data.RemoteIPAddress, data.RemotePort, data.Content.ToString());
}
break;
}
case MessageType.LoginResponse:
{
if (LoginResponse != null)
{
LoginResponse(data.RemoteIPAddress, data.RemotePort, data.Content.ToString());
}
break;
}
case MessageType.TextMessage:
{
if (TextMessageReceived != null)
{
TextMessageReceived(data.RemoteIPAddress, data.RemotePort, data.Content.ToString());
}
break;
}
case MessageType.TextMessageResponse:
{
if(TextMessageResponse != null)
{
TextMessageResponse(data.RemoteIPAddress, data.RemotePort, data.Content.ToString());
}
break;
}
case MessageType.Logout:
{
if (UserLogout != null)
{
UserLogout(data.RemoteIPAddress, data.RemotePort, data.Content.ToString());
}
break;
}
case MessageType.FileTransferRequest:
{
if (FileTransferRequest != null)
{
object[] parameters = data.Content as object[];
int transferId = (int)parameters[0];
long fileSize = (long)parameters[1];
string fileName = parameters[2].ToString();
FileTransferRequest(data.RemoteIPAddress, data.RemotePort, fileName, fileSize, transferId);
}
break;
}
case MessageType.FileTransferResponse:
{
if (FileTransferResponse != null)
{
object[] parameters = data.Content as object[];
int transferId = (int)parameters[0];
bool accepted = (bool)parameters[1];
FileTransferResponse(data.RemoteIPAddress, data.RemotePort, accepted, transferId);
}
break;
}
case MessageType.FileData:
{
if (FileDataReceived != null)
{
object[] parameters = data.Content as object[];
int transferId = (int)parameters[0];
int segmentId = (int)parameters[1];
byte[] fileData = parameters[2] as byte[];
FileDataReceived(data.RemoteIPAddress, data.RemotePort, fileData, transferId, segmentId);
}
break;
}
case MessageType.FileTransferComplete:
{
if (FileTransferComplete != null)
{
int transferId = (int)data.Content;
FileTransferComplete(data.RemoteIPAddress, data.RemotePort, transferId);
}
break;
}
case MessageType.FileTransferStopped:
{
if (FileTransferStopped != null)
{
int transferId = (int)data.Content;
FileTransferStopped(data.RemoteIPAddress, data.RemotePort, transferId);
}
break;
}
case MessageType.FileDataAcknowledgment:
{
if (FileDataAcknowledgment != null)
{
object[] parameters = data.Content as object[];
int transferId = (int)parameters[0];
int segmentId = (int)parameters[1];
FileDataAcknowledgment(data.RemoteIPAddress, data.RemotePort, transferId, segmentId);
}
break;
}
default:
{
base.ProcessData(data); // Let base class handle other cases
break;
}
}
}
}
public delegate void UserLoginEventHandler(string remoteIP, int remotePort, string userName);
public delegate void LoginResponseEventHandler(string remoteIP, int remotePort, string responseMessage);
public delegate void TextMessageReceivedEventHandler(string remoteIP, int remotePort, string message);
public delegate void TextMessageResponseEventHandler(string remoteIP, int remotePort, string response);
public delegate void UserLogoutEventHandler(string remoteIP, int remotePort, string userName);
public delegate void FileTransferRequestEventHandler(string remoteIP, int remotePort, string fileName, long fileSize, int transferId);
public delegate void FileTransferResponseEventHandler(string remoteIP, int remotePort, bool accepted, int transferId);
public delegate void FileDataReceivedEventHandler(string remoteIP, int remotePort, byte[] data, int transferId, int segmentId);
public delegate void FileTransferCompleteEventHandler(string remoteIP, int remotePort, int transferId);
public delegate void FileTransferStoppedEventHandler(string remoteIP, int remotePort, int transferId);
public delegate void FileDataAcknowledgmentEventHandler(string remoteIP, int remotePort, int transferId, int segmentId);
/// <summary>
/// Custom data analysis class that overrides the Analyse method
/// Responsible for analyzing specific data types, delegating others to the base class
/// Can derive new classes from MyDataAnalyse and override Analyse method further
/// </summary>
public class MyDataAnalyse : DataAnalyse
{
public MyDataAnalyse(UDPSocket socket)
: base(socket)
{
}
protected override void Analyze(RawData rawData)
{
string remoteIP = ((IPEndPoint)rawData.RemoteEndpoint).Address.ToString();
int remotePort = ((IPEndPoint)rawData.RemoteEndpoint).Port;
if (rawData.Buffer.Length > 0)
{
MessageType messageType = (MessageType)rawData.Buffer[0]; // Message header
switch (messageType)
{
case MessageType.Login: // Login request
{
string userData = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);
NotifyAnalysisComplete(messageType, userData, remoteIP, remotePort);
break;
}
case MessageType.LoginResponse: // Login response
{
string response = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);
NotifyAnalysisComplete(messageType, response, remoteIP, remotePort);
break;
}
case MessageType.TextMessage: // Text message
{
string message = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);
NotifyAnalysisComplete(messageType, message, remoteIP, remotePort);
break;
}
case MessageType.TextMessageResponse: // Text message response
{
string response = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);
NotifyAnalysisComplete(messageType, response, remoteIP, remotePort);
break;
}
case MessageType.Logout: // Logout notification
{
string logoutInfo = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);
NotifyAnalysisComplete(messageType, logoutInfo, remoteIP, remotePort);
break;
}
case MessageType.FileTransferRequest: // File transfer request
{
int transferId = BitConverter.ToInt32(rawData.Buffer, 1);
long fileSize = BitConverter.ToInt64(rawData.Buffer, 5);
string fileName = Encoding.Unicode.GetString(rawData.Buffer, 13, rawData.Length - 13);
NotifyAnalysisComplete(messageType, new object[] { transferId, fileSize, fileName }, remoteIP, remotePort);
break;
}
case MessageType.FileTransferResponse: // File transfer response
{
int transferId = BitConverter.ToInt32(rawData.Buffer, 1);
bool accepted = BitConverter.ToBoolean(rawData.Buffer, 5);
NotifyAnalysisComplete(messageType, new object[] { transferId, accepted }, remoteIP, remotePort);
break;
}
case MessageType.FileData: // File data segment
{
int transferId = BitConverter.ToInt32(rawData.Buffer, 1);
int segmentId = BitConverter.ToInt32(rawData.Buffer, 5);
byte[] fileSegment = new byte[rawData.Length - 9];
Buffer.BlockCopy(rawData.Buffer, 9, fileSegment, 0, fileSegment.Length);
NotifyAnalysisComplete(messageType, new object[] { transferId, segmentId, fileSegment }, remoteIP, remotePort);
break;
}
case MessageType.FileTransferComplete: // File transfer complete
{
int transferId = BitConverter.ToInt32(rawData.Buffer, 1);
NotifyAnalysisComplete(messageType, transferId, remoteIP, remotePort);
break;
}
case MessageType.FileTransferStopped: // Sender stopped transfer
{
int transferId = BitConverter.ToInt32(rawData.Buffer, 1);
NotifyAnalysisComplete(messageType, transferId, remoteIP, remotePort);
break;
}
case MessageType.FileDataAcknowledgment: // File data acknowledgment
{
int transferId = BitConverter.ToInt32(rawData.Buffer, 1);
int segmentId = BitConverter.ToInt32(rawData.Buffer, 5);
NotifyAnalysisComplete(messageType, new object[] { transferId, segmentId }, remoteIP, remotePort);
break;
}
default:
{
base.Analyze(rawData); // Let base class handle other cases
break;
}
}
}
}
}
As shown in the code, after downloading the source code, you can encapsulate UDPSocket, UDPSender, DataDeal, DataAnalyse, and types from Helper.cs into a common assembly. In actual use, you only need to reference this assembly and implement your own DataDeal and DataAnalyse classes.
Additionally, I've defined a custom "communication protocol" for this demo:
1 LoginRequest: User comes online Header + User Information
2 LoginResponse: Login response Header + Response User Information
3 TextMessage: Text message Header + Message Content
4 TextMessageResponse: Text message response Header + Original Message
5 Logout: User disconnects Header + Any text
6 FileTransferRequest: Request to send file Header + File Transfer ID + Size + Filename
7 FileTransferResponse: File transfer response Header + Transfer ID + Accept/Deny (true/false)
8 FileData: File data segment Header + Transfer ID + Segment ID + File Data (byte[])
9 FileTransferComplete: File transfer complete Header + Transfer ID
10 FileTransferStopped: Sender stopped transfer Header + Transfer ID
11 FileTransferStopped: Receiver stopped transfer Header + Transfer ID
12 FileDataAcknowledgment: File data ack Header + Transfer ID + Segment ID
Communication protocols vary based on specific requirements.
To summarize, the steps for implementing this architecture in other projects are:
- Define a "message protocol" based on specific requirements;
- Derive a new class from DataAnalyse and override the Analyse method according to the "message protocol" for data analysis;
- Derive a new class from DataDeal and overrdie the Deal method according to business logic for data processing;
- The data sender must strictly follow the "message protocol" when sending data;
- Finally, register events from the DataDeal derived class and use the data passed by the "pump."