TCP Socket exchanging kryo serialized objects

Question

I am currently working on a project for sending/receiving objects through TCP sockets. I chose to work with the Kryo Serialization framework, since it is one of the most popular frameworks out there. I saw that for Network communication, KryoNet is recommended, but for my own reasons, I chose to use my own TCP Socket framework (mainly because I want custom control of TCP streams and Threads on my project). The problem I encounter is that I have created the following class for Messages:

public class MyMessage {

    public HashMap<String, String> _values;

    public MyMessage() {
        _values = new HashMap<String, String>();
    }
}

And on the server side I have the following code for reading input from a TCP Socket:

import java.io.*;
import java.net.*;

public class MyServer {

    private ServerSocket _serverSocket;

    private Integer _serverPort;

    private boolean _killCommand;

    public MyServer() {
        _serverSocket = null;
        _serverPort = -1;
        try {
            _serverSocket = new ServerSocket(0);
            _serverPort = _serverSocket.getLocalPort();
            System.out.println("server started on IP: " + _serverSocket.getInetAddress().getHostAddress() + ":" + _serverPort);
        } catch (IOException e) {
            e.printStackTrace();
        }
        _killCommand = false;
    }

    public void runServer() {
        Socket _client= null;
        while(_killCommand == false) {
            try {
                _client = _serverSocket.accept();
                _out = _client.getOutputStream();
                _in = _client.getInputStream();
                (new Thread(new ServerThread(_in, _out))).start();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

the ServerThread class:

import java.io.*;
import java.util.HashMap;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.*;

public class ServerThread implements Runnable {

    private InputStream _in;

    private OutputStream _out;

    private Kryo kryo;

    private Input _input;

    private Output _output;

    public SynEFOthread(InputStream in, OutputStream out) {
        _in = in;
        _out = out;
        kryo = new Kryo();
        _output = new Output(_out);
        _input = new Input(_in);
    }

    @Override
    public void run() {
        MyMessage msg = null;
        System.out.println("Thread worker: about to parse input message");
        msg = kryo.readObject(_input, MyMessage.class);
        }
    }
}

On the other side, the client code is the following:

import java.io.*;
import java.net.Socket;
import java.util.*;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.*;

public class MyClient {

    public static void main(String[] args) throws Exception {
        String server_ip = "";
        Integer server_port = -1;

        server_ip = args[0];
        server_port = Integer.parseInt(args[1]);

        Socket serverSocket = new Socket(server_ip, server_port);
        OutputStream _out = serverSocket.getOutputStream();
        InputStream _in = serverSocket.getInputStream();
        Output _output = new Output(_out);
        Input _input = new Input(_in);
        MyMessage msg = new MyMessage();
        msg._values = new HashMap<String, String>();
        msg._values.put("TASK_TYPE", "TOPOLOGY");
        Kryo kryo = new Kryo();
        kryo.writeObject(_output, msg);
        Thread.sleep(100);
        kryo.writeObject(_output, _msg);
        String _ack = kryo.readObject(_input, String.class);
        serverSocket.close();
    }
}

The problem is that on ServerThread.run() function, the kryo.readObject() call blocks and nothing is done after that. Am I doing something wrong? Am I opening the streams properly for use with the Kryo Serialization Framework?

Answer 1

1. Register your class with kryo.register(MyMessage.class) both on client and server sides.
2. Create a test for your serialization/deserialization code, but instead of sockets use ByteArrayOutputStream/ByteArrayInputStream.
3. Flush! After you write complete message to stream, flush written bytes to network: _output.flush()