Differences

This shows you the differences between two versions of the page.

--- tanszek:oktatas:iss_t:java_example_for_blocking_and_non-blocking_socket [2023/03/05 16:35] – létrehozva knehez
+++ tanszek:oktatas:iss_t:java_example_for_blocking_and_non-blocking_socket [2024/03/04 08:00] (current) – [Blocking UDP sockets in Java] knehez
@@ Line 1: / Line 1: @@
-==== Blocking TCP based sockets in Java ====
+==== Blocking TCP sockets in Java ====
-=== Socket server source code ===
+=== Blocking Socket server source code ===
-<code java>
+<sxh java>
+    package org.ait;
     import java.io.IOException;
     import java.io.ObjectInputStream;
@@ Line 22: / Line 23: @@
     		try {
     			// 1. create a socket server listening to port 8080
-    			providerSocket = new ServerSocket(8080, 10);
+    			providerSocket = new ServerSocket(8080);
     			// 2. waiting for the connection (here we are waiting until next connection)
     			connection = providerSocket.accept();
@@ Line 71: / Line 72: @@
     	}
     }
-</code>
+</sxh>
-=== Socket client source ===
+=== Blocking Socket client source ===
-<code java>
+<sxh java>
+    package org.ait;
     import java.io.IOException;
     import java.io.ObjectInputStream;
@@ Line 137: / Line 139: @@
     	}
     }
-</code>
+</sxh>
-==== Non-blocking TCP based sockets in Java ====
+==== Blocking UDP sockets in Java ====
 The following Agent sends a message and waits for a response on port 8080, also with UDP. Older versions of the Eclipse IDE, the text you type on the console can be sent by pressing ctrl+z
-<code java>
+<sxh java>
-  package org.ait;
+package org.ait;
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.InetAddress;
+public class UDPClient {
+    public static void main(String args[]) throws Exception {
+        BufferedReader inFromUser = new BufferedReader(new InputStreamReader(System.in));
+        DatagramSocket clientSocket = new DatagramSocket();
+        InetAddress IPAddress = InetAddress.getByName("localhost");
+        byte[] sendData = new byte[1024];
+        byte[] receiveData = new byte[1024];
+        String sentence = inFromUser.readLine();
+        sendData = sentence.getBytes();
+        DatagramPacket sendPacket = new DatagramPacket(sendData, sendData.length, IPAddress, 8080);
+        clientSocket.send(sendPacket);
+        DatagramPacket receivePacket = new DatagramPacket(receiveData, receiveData.length);
+        clientSocket.receive(receivePacket);
+        String modifiedSentence = new String(receivePacket.getData());
+        System.out.println("converted:" + modifiedSentence);
+        clientSocket.close();
+    }
+}
+</sxh>
+The UDP server waits for the agents messages on port 8080 and converts them to uppercase letters and sends them back to the client UDP socket.
+<sxh java>
+package org.ait;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.InetAddress;
+public class UDPServer {
+    public static void main(String args[]) throws Exception {
+        DatagramSocket serverSocket = new DatagramSocket(8080);
+        byte[] bytesReceived = new byte[1024];
+        byte[] bytesSent = new byte[1024];
+        DatagramPacket receivePacket = new DatagramPacket(bytesReceived, bytesReceived.length);
+        // here we are waiting for the packets
+        serverSocket.receive(receivePacket);
+        String textMessage = new String(receivePacket.getData());
+        System.out.println("I got: " + textMessage);
+        InetAddress IPAddress = receivePacket.getAddress();
+        int port = receivePacket.getPort();
+        String upperCaseText = textMessage.toUpperCase();
+        bytesSent = upperCaseText.getBytes();
+        // send back
+        DatagramPacket sendPacket = new DatagramPacket(bytesSent, bytesSent.length, IPAddress, port);
+        serverSocket.send(sendPacket);
+        serverSocket.close();
+    }
+}
+</sxh>
+==== Non-blocking TCP sockets in Java ====
+{{:tanszek:oktatas:iss_t:sockets-blocking-nonblocking.png?800|}}
+Reading:
+  * http://tutorials.jenkov.com/java-nio/nio-vs-io.html
+  * http://www.javaworld.com/article/2073344/core-java/use-select-for-high-speed-networking.html
+{{tanszek:oktatas:informacios_rendszerek_integralasa:high_speed_socket.jpg|}}
+===== Non-blocking loop =====
+<sxh java>
+   ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
+   serverSocketChannel.socket().bind(new InetSocketAddress(9999));
+   serverSocketChannel.configureBlocking(false);  // ez a sor jelzi a blokkolásmentes működést
+   while(true){
+    SocketChannel socketChannel = serverSocketChannel.accept();
+    if(socketChannel != null){
+        // the connection is accepted
+    }
+   }
+</sxh>
+===== Non-blocking Java client example =====
+<sxh java>
+  import java.io.IOException;
+  import java.io.InputStream;
+  import java.io.OutputStream;
+  import java.net.Socket;
+  import java.util.Random;
-  import java.io.BufferedReader;
+  public class Client implements Runnable {
-  import java.io.InputStreamReader;
+	private String host;
-  import java.net.DatagramPacket;
+	private int port;
-  import java.net.DatagramSocket;
-  import java.net.InetAddress;
-  public class UDPClient {
+	// Bounds on how much we write per cycle
-	public static void main(String args[]) throws Exception {
+	private static final int minWriteSize = 1024;
-		BufferedReader inFromUser = new BufferedReader(new InputStreamReader(System.in));
+	private static final int maxWriteSize = 65536;
-		DatagramSocket clientSocket = new DatagramSocket();
-		InetAddress IPAddress = InetAddress.getByName("localhost");
+	// Bounds on how long we wait between cycles
+	private static final int minPause = (int) (0.05 * 1000);
-		byte[] sendData = new byte[1024];
+	private static final int maxPause = (int) (0.5 * 1000);
-		byte[] receiveData = new byte[1024];
+	// Random number generator
-		String sentence = inFromUser.readLine();
+	Random rand = new Random();
-		sendData = sentence.getBytes();
+	public Client(String host, int port, int numThreads) {
-		DatagramPacket sendPacket = new DatagramPacket(sendData, sendData.length, IPAddress, 8080);
+		this.host = host;
-		clientSocket.send(sendPacket);
+		this.port = port;
-		DatagramPacket receivePacket = new DatagramPacket(receiveData, receiveData.length);
+		for (int i = 0; i < numThreads; ++i) {
-		clientSocket.receive(receivePacket);
+			new Thread(this).start();
-		String modifiedSentence = new String(receivePacket.getData());
+		}
-		System.out.println("converted:" + modifiedSentence);
-		clientSocket.close();
 	}
+	public void run() {
+		byte buffer[] = new byte[maxWriteSize];
+		try {
+			Socket s = new Socket(host, port);
+			InputStream in = s.getInputStream();
+			OutputStream out = s.getOutputStream();
+			while (true) {
+				int numToWrite = minWriteSize
+						+ (int) (rand.nextDouble() * (maxWriteSize - minWriteSize));
+				for (int i = 0; i < numToWrite; ++i) {
+					buffer[i] = (byte) rand.nextInt(256);
+				}
+				out.write(buffer, 0, numToWrite);
+				int sofar = 0;
+				while (sofar < numToWrite) {
+					sofar += in.read(buffer, sofar, numToWrite - sofar);
+				}
+				System.out.println(Thread.currentThread() + " wrote " + numToWrite);
+				int pause = minPause + (int) (rand.nextDouble() * (maxPause - minPause));
+				try {
+					Thread.sleep(pause);
+				} catch (InterruptedException ie) {
+				}
+			}
+		} catch (IOException ie) {
+			ie.printStackTrace();
+		}
 	}
-</code>
+	static public void main(String args[]) throws Exception {
+		new Client("localhost", 4444, 3);
+	}
+  }
-The UDP server waits for the agents messages on port 8080 and converts them to uppercase letters and sends them back to the client UDP socket.
+</sxh>
-<code java>
-  package org.ait;
+===== Non-blocking Java server example =====
+<sxh java>
-  import java.net.DatagramPacket;
+  public class Server implements Runnable {
-  import java.net.DatagramSocket;
+	// The port we will listen on
-  import java.net.InetAddress;
+	private int port;
+	// A pre-allocated buffer for encrypting data
-  public class UDPServer {
+	private final ByteBuffer buffer = ByteBuffer.allocate(16384);
-	public static void main(String args[]) throws Exception {
+	public Server(int port) {
+		this.port = port;
-		DatagramSocket serverSocket = new DatagramSocket(8080);
+		new Thread(this).start();
+	}
-		byte[] bytesReceived = new byte[1024];
+	public void run() {
-		byte[] bytesSent = new byte[1024];
+		try {
+			// Instead of creating a ServerSocket,
+			// create a ServerSocketChannel
-		DatagramPacket receivePacket = new DatagramPacket(bytesReceived, bytesReceived.length);
+			ServerSocketChannel ssc = ServerSocketChannel.open();
-		// here we are waiting for the packets
+			// Set it to non-blocking, so we can use select
-		serverSocket.receive(receivePacket);
+			ssc.configureBlocking(false);
+			// Get the Socket connected to this channel, and bind it
-		String textMessage = new String(receivePacket.getData());
+			// to the listening port
+			ServerSocket ss = ssc.socket();
-		System.out.println("I got: " + textMessage);
+			InetSocketAddress isa = new InetSocketAddress(port);
+			ss.bind(isa);
-		InetAddress IPAddress = receivePacket.getAddress();
+			// Create a new Selector for selecting
-		int port = receivePacket.getPort();
+			Selector selector = Selector.open();
+			// Register the ServerSocketChannel, so we can
-		String upperCaseText = textMessage.toUpperCase();
+			// listen for incoming connections
-		bytesSent = upperCaseText.getBytes();
+			ssc.register(selector, SelectionKey.OP_ACCEPT);
+			System.out.println("Listening on port " + port);
-		// send back
+			while (true) {
-		DatagramPacket sendPacket = new DatagramPacket(bytesSent, bytesSent.length, IPAddress, port);
+				// See if we've had any activity -- either
-		serverSocket.send(sendPacket);
+				// an incoming connection, or incoming data on an
-		serverSocket.close();
+				// existing connection
+				int num = selector.select();
+				// If we don't have any activity, loop around and wait
+				// again
+				if (num == 0) {
+					continue;
+				}
+				// Get the keys corresponding to the activity
+				// that has been detected, and process them
+				// one by one
+				Set keys = selector.selectedKeys();
+				Iterator it = keys.iterator();
+				while (it.hasNext()) {
+					// Get a key representing one of bits of I/O
+					// activity
+					SelectionKey key = (SelectionKey) it.next();
+					// What kind of activity is it?
+					if ((key.readyOps() & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT) {
+						System.out.println("acc");
+						// It's an incoming connection.
+						// Register this socket with the Selector
+						// so we can listen for input on it
+						Socket s = ss.accept();
+						System.out.println("Got connection from " + s);
+						// Make sure to make it non-blocking, so we can
+						// use a selector on it.
+						SocketChannel sc = s.getChannel();
+						sc.configureBlocking(false);
+						// Register it with the selector, for reading
+						sc.register(selector, SelectionKey.OP_READ);
+					} else if ((key.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {
+						SocketChannel sc = null;
+						try {
+							// It's incoming data on a connection, so
+							// process it
+							sc = (SocketChannel) key.channel();
+							boolean ok = processInput(sc);
+							// If the connection is dead, then remove it
+							// from the selector and close it
+							if (!ok) {
+								key.cancel();
+								Socket s = null;
+								try {
+									s = sc.socket();
+									s.close();
+								} catch (IOException ie) {
+									System.err.println("Error closing socket "
+											+ s + ": " + ie);
+								}
+							}
+						} catch (IOException ie) {
+							// On exception, remove this channel from the
+							// selector
+							key.cancel();
+							try {
+								sc.close();
+							} catch (IOException ie2) {
+								System.out.println(ie2);
+							}
+							System.out.println("Closed " + sc);
+						}
+					}
+				}
+				// We remove the selected keys, because we've dealt
+				// with them.
+				keys.clear();
+			}
+		} catch (IOException ie) {
+			System.err.println(ie);
+		}
+	}
+	// Do some cheesy encryption on the incoming data,
+	// and send it back out
+	private boolean processInput(SocketChannel sc) throws IOException {
+		buffer.clear();
+		sc.read(buffer);
+		buffer.flip();
+		// If no data, close the connection
+		if (buffer.limit() == 0) {
+			return false;
+		}
+		// Simple rot-13 encryption
+		for (int i = 0; i < buffer.limit(); ++i) {
+			byte b = buffer.get(i);
+			if ((b >= 'a' && b <= 'm') || (b >= 'A' && b <= 'M')) {
+				b += 13;
+			} else if ((b >= 'n' && b <= 'z') || (b >= 'N' && b <= 'Z')) {
+				b -= 13;
+			}
+			buffer.put(i, b);
+		}
+		sc.write(buffer);
+		System.out.println("Processed " + buffer.limit() + " from " + sc);
+		return true;
+	}
+	static public void main(String args[]) throws Exception {
+		new Server(4444);
 	}
   }
-</code>
+</sxh>