2
\$\begingroup\$

I write kind of UDP server. It should serves about 50 messages per second. I would like a review about the general code 'quality' and logic. The server uses the same socket for sending and receiving messages. The messages are short - about 100 bytes (json).

The code:

public class UdpServer extends Thread {

/* Receive timeout in mills */
private static int RCV_TIMEOUT = 100;       

/* Buffer size */
final private static int BUFF_SIZE = 512;           

/* Number of concurrent handlers */ 
final private static int THREAD_POOL_SIZE = 10;      


/* Socket */
final private DatagramSocket socket;

/* Thread pool executor */
final private ThreadPoolExecutor executor;

/* Termination states */
private boolean terminated = false;     //set to true if this server is terminated
private boolean terminating = false;    //set to true if this server is in the process of terminating   

/* 
 * Debug flag. If it is ON then server sends 
 * debug messages to clients 
 */
private final static boolean debugModeOn = false;

/*
 * Transmitter to use by handler to send response to request 
 * originator. It encapsulates socket, so handler may only send data.
 */
private DatagramResponder responder;


public UdpServer (String ip, int port) 
throws ValidationException, SocketException {

    if (port <= 0)
        throw new ValidationException("Illegal port: " + port);

    InetAddressValidator validator = InetAddressValidator.getInstance();

    if (!validator.isValidInet4Address(ip) && !validator.isValidInet6Address(ip))
        throw new ValidationException("Illegal Ip address: " + ip);

    if (!Utils.isIpLocalAndUp(ip))
        throw new ValidationException("Given IP is not a local IP or interface is down");

    SocketAddress socketAddress = new InetSocketAddress(ip, port);
    socket = new DatagramSocket(socketAddress);
    socket.setSoTimeout(RCV_TIMEOUT);

    executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(THREAD_POOL_SIZE);

    responder = new Responder (socket);
}



public void run()  {

    while (!terminated) {

        byte[] buf = new byte[BUFF_SIZE];

        DatagramPacket packet = new DatagramPacket(buf, buf.length);

        try {
            socket.receive(packet);
        } catch (SocketTimeoutException to) {

            if (!terminated) continue;
            else break;

        } catch (Exception e) {
            e.printStackTrace();
            return;
        }

        if (!terminating) {
            Runnable handler = HandlerFactory.getHandler(packet, responder);
            executor.execute(handler);
        }

    }

    socket.close();
}


/**
 * @return true if server debug mode is ON
 */
public static boolean isDebugModeOn() {
    return debugModeOn;
}

/*
 * Stops the server. The method waits (up to 60 seconds) 
 * until all tasks have completed execution 
 */
public void terminate () throws IOException {

    if (terminating || terminated)
        return;

    terminating = true;

    try {
        executor.awaitTermination(60, TimeUnit.SECONDS);
    } catch (InterruptedException e) {};

    executor.shutdownNow();
    terminated = true;  
}


//for resting only, will be removed
private static void setReceiveTO (int timeOut) {
    RCV_TIMEOUT = timeOut;
}



private static class Responder implements DatagramResponder {


    private DatagramSocket socket;

    private Responder (DatagramSocket socket) {
        this.socket = socket;
    }

    @Override
    public void send(DatagramPacket packet) throws IOException, SecurityException, 
                PortUnreachableException, IllegalBlockingModeException, IllegalArgumentException {

        socket.send(packet);

    }

}


public static void main(String[] args) throws ValidationException, IOException, InterruptedException {

    //Test that we can send data while socket is blocked on receive
    UdpServer.setReceiveTO(1000 * 60 * 60); //1 hour
    UdpServer server = new UdpServer ("127.0.0.1", 1234);
    server.start();


    TestReceiver rcv = new TestReceiver();
    rcv.start();

    TestSender tester = new TestSender(server.socket);
    tester.start();
}


//Receiver 
private static class TestReceiver extends Thread {


    private DatagramSocket socket;

    private TestReceiver () throws SocketException {
        socket = new DatagramSocket(1000);
    }

    public void run() {

        while (true) {

            try {
                byte [] buf = new byte[100];
                DatagramPacket packet = new DatagramPacket(buf, buf.length);
                socket.receive(packet);
                System.out.println("Received: " + new String (packet.getData()));
            } catch (IOException e) {
                System.out.println (e.getMessage());
            }
        }
    }
}



private static class TestSender extends Thread {

    private DatagramSocket socket;

    private byte [] buf = "hello".getBytes();
    private int port = 1000;
    private InetSocketAddress address = new InetSocketAddress("127.0.0.1", port);

    private DatagramPacket packet = new DatagramPacket(buf, buf.length, address.getAddress(), port);

    private TestSender (DatagramSocket socket) {
        this.socket = socket;
    }

    public void run() {

        while (true) {
            try {
                socket.send(packet);
            } catch (IOException e) {
                System.out.println (e.getMessage()); 
            }
        }
    }
}

}

\$\endgroup\$
  • \$\begingroup\$ Your code ends with .... Is it complete? \$\endgroup\$ – JAD Sep 9 at 9:47
  • \$\begingroup\$ Yes it is complete. Rest of code is irrelavant (test cases and main method), so I cut it \$\endgroup\$ – bw_dev Sep 9 at 10:04
  • \$\begingroup\$ @bw_dev If you have test cases add them to the question too. Include main method as well. this way we can take a look at how this class is used / tested. \$\endgroup\$ – bhathiya-perera Sep 9 at 10:15
  • \$\begingroup\$ Code updated - added main and two test classes (will be removed from final version) \$\endgroup\$ – bw_dev Sep 9 at 11:48
  • \$\begingroup\$ What is the version of Java? \$\endgroup\$ – bhathiya-perera Sep 9 at 12:14
4
\$\begingroup\$

General

Classes should be final unless you intend and have designed them to be extended. Marking local variables as final will clue the reader in that they don't change once assigned, which improves readability.

Use full, descriptive variable names. The extra characters don't cost you anything, and they make the code more clear and easier to read.

The suggested keyword order is private static final, not final private static.

Most of your comments are noise.

terminating and terminated are false by default and don't need to be explicitly assigned.

In idiomatic java,

  • there is no whitespace between the end of method name and the `(`.
  • there is no whitespace between a constructor name and the `(`.
  • there is no whitespace after a variable type declaration and before the `[]`.
  • curly braces are always used, even when they are optional. This makes your code consistent and removes a common class of error when refactoring existing code.
  • only constants use UPPERCASE, and all constants should use UPPERCASE.

UdpServer

You should be instantiating Runnable, not extending Thread. You have not created a generally-useful thread that adds new functionality, but rather a set of functionality that happens to want to run on a thread.

Adding testing-specific methods is usually a bad idea. In this case, it would be very reasonable for the timeout to be an optional constructor parameter, since clients may very well want a non-default value set. Your test client certainly does.

Try to provide meaningful failure messages. Don't just say something is wrong, give a hint on how to fix it.

You're performing the same check twice in the block that starts if (!validator.isValidInet6Address.

Your debug-mode code doesn't do anything and should be removed. Generally, code shouldn't have this kind of property. You're basically compiling in how much information the client is getting, rather than making it configurable in some way.

run()

Sockets should always be closed. This is not the case in your code, where an exception can cause the socket to remain open. Move socket instantiation to run and put it in a try-with-resources block.

It seems to me that you want to stop accepting new data on terminating, not terminated. I don't know your requirements, but at a high level it seems wrong to accept new requests while you're trying to shut down the server. If you do make that logical change, then you can switch from two variables to one, running, which doesn't need to be negated everywhere it's used.

You don't need to check for termination in your catch of SocketTimeoutException. Just continue and the while loop condition will handle the check for you.

Always catch the most specific exception you know how to handle. Catching Exception is a bad idea because it will automatically handle any checked exceptions that get added your try block later, whether that was your intent or not.

terminate()

This method doesn't actually throw an IOException. You should turn on compiler warnings aggressively to help notice mistakes like this.

It's probably a better idea to shutdownNow() if somebody sends an interrupt. If you feel strongly against it, inside your catch block you should put Thread.currentThread().interrupt(). This preserves the interrupted status, rather than consuming it.

Responder

Responder's constructor should be public. You've declared a class that can only be seen by its parent and who can't be constructed by anybody outside the class. You want a class that can only be seen by its parent but can be constructed by anybody who can see it.

I don't know what a DatagramResponder is, but declaring all those thrown exception types in your send() signature is probably not idea. It's typical to only include checked exceptions in the method signature, and to document in your JavaDoc what unchecked exceptions might be thrown. Typically unchecked exceptions can't be reasonably handled by callers. It also suggests that the send method might be doing too much, if it can throw so many kinds of exception.


I'm not going to review the test code, except to say that if you really need to expose the UdpServer socket for testing, you can make it a private instance variable that gets assigned in run(). You can then expose it with a package-private accessor method. I'd strongly advise against testing this way, though. Test UdpServer by actually running UdpServer.


If you were to make all these changes, your code might look something like:

public final class UdpServer implements Runnable {

    private static final int BUFFER_SIZE = 512;

    private static final int CONCURRENT_HANDLERS = 10;

    private final int receiveTimeoutMillis;

    private final SocketAddress socketAddress;

    private final ThreadPoolExecutor executor;

    private boolean running;

    public UdpServer(final String ip, final int port)
            throws ValidationException {
        this(ip, port, 100);
    }

    public UdpServer(final String ip, final int port, final int receiveTimeoutMillis)
            throws ValidationException {

        if (receiveTimeoutMillis <= 0) {
            throw new ValidationException("Timeout must be greater than zero, was: " + receiveTimeoutMillis);
        }
        this.receiveTimeoutMillis = receiveTimeoutMillis;

        if (port <= 0) {
            throw new ValidationException("Port must be greater than zero, was: " + port);
        }

        final InetAddressValidator validator = InetAddressValidator.getInstance();
        if (!validator.isValidInet6Address(ip)) {
            throw new ValidationException("Illegal Ip address: " + ip);
        }

        if (!Utils.isIpLocalAndUp(ip)) {
            throw new ValidationException("Given IP is not a local IP or interface is down");
        }

        this.socketAddress = new InetSocketAddress(ip, port);
        this.executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(CONCURRENT_HANDLERS);
    }


    @Override
    public void run()  {
        this.running = true;

        try (final DatagramSocket socket = new DatagramSocket(this.socketAddress)) {
            socket.setSoTimeout(this.receiveTimeoutMillis);
            final Responder responder = new Responder(socket);

            while (this.running) {
                final byte[] buffer = new byte[BUFFER_SIZE];
                final DatagramPacket packet = new DatagramPacket(buffer, buffer.length);

                try {
                    socket.receive(packet);
                } catch (final SocketTimeoutException e) {
                    continue;
                } catch (final IOException e) {
                    System.err.println("Error reading from socket");
                    e.printStackTrace();
                    return;
                }

                if (this.running) {
                    final Runnable handler = HandlerFactory.getHandler(packet, responder);
                    this.executor.execute(handler);
                }
            }
        } catch (final SocketException e) {
            System.err.println("Error while opening socket");
            e.printStackTrace();
        }
    }

    /**
     * Stops the server. The method waits (up to 60 seconds)
     * until all tasks have completed execution unless interrupted.
     */
    public void terminate() {
        if (!this.running) {
            return;
        }

        this.running = false;

        try {
            this.executor.awaitTermination(60, TimeUnit.SECONDS);
            this.executor.shutdownNow();
        } catch (final InterruptedException e) {
            this.executor.shutdownNow();
            // or Thread.currentThread().interrupt();
        }
    }


    /*
     * Transmitter to use by handler to send response to request
     * originator. It encapsulates socket, so handler may only send data.
     */
    private static final class Responder implements DatagramResponder {

        private final DatagramSocket socket;

        public Responder(final DatagramSocket socket) {
            this.socket = socket;
        }

        @Override
        public void send(final DatagramPacket packet)
                throws IOException, PortUnreachableException, IllegalBlockingModeException {
            this.socket.send(packet);
        }

    }
}
\$\endgroup\$
  • \$\begingroup\$ Thanks for the comment. The only question here: in your implementation server may exit before handlers finished their work. The server thread is "parent" of handler threads. May this cause handlers be destroyed without finishing tasks? \$\endgroup\$ – bw_dev Sep 9 at 17:01
  • \$\begingroup\$ @bw_dev Ah. If you want the handlers to finish, then yes, two variables is a better approach. It depends on what semantics you want for terminate(). \$\endgroup\$ – Eric Stein Sep 9 at 19:51

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.