带有序列号的 UDP

2022-01-22 00:00:00 sockets client-server udp java

我正在尝试为 Java 中的类分配实现可靠的 UDP 协议.我已经设法将确认添加到接收到的每个数据报包中，但是在我发送的数据报包中实现序列号时遇到了麻烦.

I am trying to implement a Reliable UDP protocol for a class assignment in Java. I have managed to add the acknowledgments to every datagram packet that is received, but I am having trouble implementing Sequence Numbers in the datagram packets that I am sending.

谁能推荐一个简单的方法来实现它?

Can anyone suggest an easy method to implement this?

@EJP 我已尝试实施您刚才的建议.这是我到目前为止的代码(它仍然非常原始 - 我使用 hit and try 方法来实现它)

@EJP I have tried implementing what you just suggested. This is my code till now (its still very raw - i was using hit and try method to implement it)

服务器端

    public class TestServer extends Activity {

private DatagramSocket serverSocket;
Thread serverThread = null;
byte[] incomingData;
byte[] outgoingData;
//int numBytesRead = 0;
int ackSent = 0;
int numPackRecv = 0;
int BUF_SIZE = 1024;
String msg = "ACK";
BufferedInputStream data=null;
BufferedOutputStream out =null;

public static final int SERVERPORT = 6000;

String outputFile = "/sdcard/Movies/asddcopy.mp4";

@Override
public void onCreate(Bundle savedInstanceState) {

    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_test_server);

    this.serverThread = new Thread(new ServerThread());
    this.serverThread.start();

}

@Override
protected void onStop() {
    super.onStop();
    try {
        serverSocket.close();
    } catch (Exception e) {
        Log.d("SERVER", "Inside onStop()");
        Log.d("SERVER", Log.getStackTraceString(e));
    }
}

class ServerThread implements Runnable {

    @SuppressLint("NewApi")
    public void run() {

        try {
            serverSocket = new DatagramSocket(SERVERPORT);
            incomingData = new byte[BUF_SIZE];
            //outgoingData = new byte[512];
            outgoingData = msg.getBytes();

            long startRxPackets = TrafficStats.getUidRxPackets(Process.myUid());
            long startTime = System.nanoTime();

            out = new BufferedOutputStream(new FileOutputStream(outputFile, true));


            while (!Thread.currentThread().isInterrupted()) {

                //serverSocket.setSoTimeout(5000);
                while (true) {
                    try{
                        //DatagramPacket incomingPacket = new DatagramPacket(incomingData, incomingData.length);
                        DatagramPacket incomingPacket = new DatagramPacket(incomingData, BUF_SIZE);

                        serverSocket.receive(incomingPacket);
                        byte[] data = incomingPacket.getData();
                        //out.write(data,0,incomingPacket.getLength());
                        //String msg = new String(incomingPacket.getData());

                        ByteArrayInputStream in = new ByteArrayInputStream(data);
                        ObjectInputStream is = new ObjectInputStream(in);

                        if (is == null) {
                            is = new ObjectInputStream(in);
                        }
                        Message msg = (Message) is.readObject();
                        System.out.println(msg.getSeqNo());
                        /*if ("END".equals(msg.substring(0, 3).trim())) {
                            Log.d("SERVER", "Inside END condition");
                            break;
                        }*/
                        out.write(msg.getData(),0,msg.getData().length);
                        numPackRecv += 1;

                        Log.d("SERVER", "Packet Received: " + numPackRecv);

                        InetAddress client = incomingPacket.getAddress();
                        int client_port = incomingPacket.getPort();
                        DatagramPacket outgoingPacket = new DatagramPacket(outgoingData, outgoingData.length, client, client_port);
                        serverSocket.send(outgoingPacket);
                        ackSent += 1;

                        //Log.d("SERVER","Packet Received: " + numPackRecv + " :: " + "Ack Sent: " + ackSent);
                    }catch(Exception e) {
                        Log.d("SERVER", "Inside run() ex1");
                        Log.d("SERVER", Log.getStackTraceString(e));
                        break;

                    }
                }

                out.close();

                serverSocket.disconnect();
                serverSocket.close();

                Log.d("SERVER", "Transfer Complete");
                 Log.d("SERVER", "Actual Time elapsed = " + (System.nanoTime() - startTime)/Math.pow(10, 9) + " s");
                 Log.d("SERVER", "Total Packets Received = " + Long.toString(TrafficStats.getUidRxPackets(Process.myUid()) - startRxPackets));
                 Log.d("SERVER", "Packets Received from Socket = " + numPackRecv);
                 break;
            }
            out.close();
            serverSocket.disconnect();
            serverSocket.close();

            /* Log.d("SERVER", "Transfer Complete");
             Log.d("SERVER", "Actual Time elapsed = " + (System.nanoTime() - startTime)/Math.pow(10, 9) + " s");
             Log.d("SERVER", "Total Packets Received = " + Long.toString(TrafficStats.getUidRxPackets(Process.myUid()) - startRxPackets));
             Log.d("SERVER", "Packets Received from Socket = " + numPackRecv);*/

        }catch (Exception e) {
            Log.d("SERVER", "Inside run() ex2");
            Log.d("SERVER", Log.getStackTraceString(e));
            serverSocket.disconnect();
            serverSocket.close();
            }
        }
    }

这是客户端

    public class TestClient extends Activity { private DatagramSocket clientSocket;
    byte[] incomingData;
    int BUF_SIZE = 500;
    int numBytesRead = 0;
    int numPackSent = 0;

    private static final int SERVERPORT = 6000;
    private static final String SERVER_IP = "10.0.0.22";

    String inFile = "/sdcard/Movies/asdd.mp4";

    @Override
    public void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_test_client);      

new Thread(new ClientThread()).start();
    }

    public void onClick(View view) {

new workInProgress().execute("");
    }

    private class workInProgress extends AsyncTask<Object, Object, Object> {

@SuppressLint("NewApi")
@Override
protected Object doInBackground(Object... params) {

    try {

        Log.d("CLIENT", "Sending a file to the server...");
        BufferedInputStream inputBuf = new BufferedInputStream(new FileInputStream(inFile));
        //byte[] fileBytes = new byte[(int) inFile.length()];
        byte[] fileBytes = new byte[BUF_SIZE];

        incomingData = new byte[BUF_SIZE];
        double numPktToSend = Math.ceil(inFile.length()*1.0/BUF_SIZE);
        //Log.d("CLIENT", "Total packets to be sent = " + numPktToSend);

        int sleepCycle = 1;
        long sysPackSent = 0;
        //long startTxPackets = TrafficStats.getTotalTxPackets();
        long startTxPackets = TrafficStats.getUidTxPackets(Process.myUid());
        Log.d("CLIENT", "startTxPacks: " + startTxPackets);
        long packDrops = 0;
        long startTime = System.nanoTime();
        long count=0;
        long ackRec=0;
        int seqNo = 0;

        ByteArrayOutputStream outStream = new ByteArrayOutputStream();
        ObjectOutputStream os = new ObjectOutputStream(outStream);


        while((numBytesRead = inputBuf.read(fileBytes)) != -1) {
            //DatagramPacket packet = new DatagramPacket(fileBytes, fileBytes.length);
             if (os == null) {
                    os = new ObjectOutputStream(outStream);
             }
             Message msg = new Message(++seqNo, fileBytes, false);   
             os.writeObject(msg);
             os.flush();
             os.reset();

             byte[] data = outStream.toByteArray();
             DatagramPacket packet = new DatagramPacket(data, data.length);
            clientSocket.send(packet);
            numPackSent += 1;

            //Log.d("CLIENT", "No of packets sent = " + numPackSent);

            sysPackSent = TrafficStats.getUidTxPackets(Process.myUid()) - startTxPackets;

            try{
                clientSocket.setSoTimeout(5000);
                packet = new DatagramPacket(incomingData, incomingData.length);
                clientSocket.receive(packet);
                String recAck = new String(packet.getData());
                ackRec++;
            }
            catch(Exception e) {
                //Log.d("CLIENT", Log.getStackTraceString(e));
            }

            packDrops = numPackSent - ackRec;
            if (packDrops > count) {
                sleepCycle = Math.min(16, sleepCycle * 2);
                count = packDrops;
                Log.d("CLIENT",String.valueOf(sleepCycle) + " :: " + numPackSent);
            } else {
                sleepCycle = Math.max(sleepCycle - 1, 1);
            }
            Thread.sleep(sleepCycle);

        }

        if (numBytesRead == -1) {
            fileBytes = "END".getBytes();
            Log.d("CLIENT", "Sending END Packet");
            clientSocket.send(new DatagramPacket(fileBytes, fileBytes.length));
        }


        Log.d("CLIENT", "Actual Time elapsed = " + (System.nanoTime() - startTime)/Math.pow(10, 9) + " s");
        Log.d("CLIENT", "Total Packets Transmitted = " + Long.toString(sysPackSent));
        Log.d("CLIENT", "No of packets dropped = " + String.valueOf(packDrops));
        Log.d("CLIENT", "Packets Pushed to Socket = " + numPackSent);
        Log.d("CLIENT", "Number of Acknoledgments received " +ackRec);

        inputBuf.close();
        os.close();
        outStream.close();
        clientSocket.disconnect();
        clientSocket.close();

        Log.d("CLIENT", "Sending file.. Complete!!!");



} catch (Exception e) {
    Log.d("CLIENT", Log.getStackTraceString(e));
    clientSocket.disconnect();
    clientSocket.close();
}

    return null;
}

    }

    class ClientThread implements Runnable  {

@Override
public void run() {

    try {
        InetAddress serverAddr = InetAddress.getByName(SERVER_IP);
        clientSocket = new DatagramSocket();
        clientSocket.connect(serverAddr, SERVERPORT);
        Log.d("CLIENT", "Connection Successful");
    } catch (UnknownHostException e1) {
        Log.d("CLIENT", "Inside run() UnknownHostEx");
        Log.d("CLIENT", Log.getStackTraceString(e1));
    } catch (IOException e1) {
        Log.d("CLIENT", "Inside run() IOEx");
        Log.d("CLIENT", Log.getStackTraceString(e1));
    }

}

    }

我在服务器端遇到了一些错误:

I am getting a few errors at the Server side:

  1. 我收到的每个数据包的序列号相同(即 1)
  2. 我不确定传入数据包的缓冲区大小，因为我在客户端使用 500 个字节，在服务器端使用 1024 个字节.如果我在两个代码中都占用 500 个字节，我会得到一个文件结束异常.

如果您能提出更好的方法来实现同样的事情，我将不胜感激！谢谢:)

I would really appreciate if you could suggest better ways to implement the same thing! Thanks :)

谢谢！

推荐答案

  1. 创建一个ByteArrayOutputStream.
  2. 将其包装在 DataOutputStream
    3. 中
  3. 使用DataOutputStream.writeInt()写入序号.
  4. 使用 write() 写入数据.
  5. 根据ByteArrayOutputStream返回的字节数组构造DatagramPacket.
  1. Create a ByteArrayOutputStream.
  2. Wrap it in a DataOutputStream
  3. Use DataOutputStream.writeInt() to write the sequence number.
  4. Use write() to write the data.
  5. Construct the DatagramPacket from the byte array returned by the ByteArrayOutputStream.

在接收端，完全相反，在每种情况下使用互补的类和方法.这些是什么留给读者作为练习.

At the receiver, do exactly the reverse, using the complementary classes and methods in each case. What those are is left as an exercise for the reader.

相关文章