Делаю лабораторную работу и очень нуждаюсь в помощи. У меня есть клиент-серверное приложение с GUI на JavaFX. От клиента приходят команды работы с коллекцией, они прекрасно работают. Проблема началась, когда я стал реализовывать оповещения, которые приходят ко всем пользователям, когда на сервере выполняются команды типа "добавить новый элемент", "удалить элемент". Вот коллекция из базы данных, которая должна передаваться. Этот скриншот сделан из debugger. В этом режиме ошибки не наблюдается.
Этот скриншот с выполнением без дебаггера.
это лог сервера, класс ServerMain.
Connection to PostgreSQL JDBC
PostgreSQL JDBC Driver successfully connected
db инициализована: jdbc:postgresql://localhost:8777/studs
18:33:11.005 [main] INFO com.itmo.server.ServerWithThreads - Запуск сервера, порт: 8080
18:33:11.243 [main] INFO com.itmo.server.ServerWithThreads - accepted client
18:33:12.850 [main] INFO com.itmo.server.ServerWithThreads - accepted client
18:33:12.873 [Thread-7] INFO com.itmo.server.ReadRequestThread - получено байт:351
18:33:12.893 [Thread-7] INFO com.itmo.server.ReadRequestThread - Полученная команда: com.itmo.commands.SubscribeForNotificationsCommand@60a89d75
я получил команду Subscribe от java.nio.channels.SocketChannel[connected local=/127.0.0.1:8080 remote=/127.0.0.1:58727]
я передал коллекцию по channel java.nio.channels.SocketChannel[connected local=/127.0.0.1:8080 remote=/127.0.0.1:58727]
я зарегистрировал подписчика java.nio.channels.SocketChannel[connected local=/127.0.0.1:8080 remote=/127.0.0.1:58727]
я выхожу из requestExecutorThread
Это то, что выводит класс для запуска GUI и подключения к серверу:
Пытаюсь установить соединение с сервером (localhost:8080)
Подключено: Socket[addr=localhost/127.0.0.1,port=8080,localport=58723]
авг 02, 2020 6:33:12 PM javafx.fxml.FXMLLoader$ValueElement processValue
WARNING: Loading FXML document with JavaFX API of version 11.0.1 by JavaFX runtime of version 8.0.241
Пытаюсь установить соединение с сервером (localhost:8080)
Подключено: Socket[addr=localhost/127.0.0.1,port=8080,localport=58727]
----- получил байты, количество: 4096
-------вот дракон из этого оповещения:
owner: admin
id: 29
name: Name
coordinates: 5, 5
creationDate: 2020-07-29
age: 100
wingspan: 8.0
type: WATER
character: CUNNING
killer: killer, birthday: 2000-01-01, haircolor: GREEN, nationality: RUSSIA, location: x: 5, y: 5, z: 1.0, locname: 5
----------notificationInBytes был равен null
----- получил байты, количество: 4096
-------вот дракон из этого оповещения:
owner: admin
id: 36
name: test
coordinates: 1, 1
creationDate: 2020-07-29
age: 5
wingspan: 5.0
type: AIR
character: CUNNING
killer: killer, birthday: 2020-07-29, haircolor: BROWN, nationality: RUSSIA, location: x: 1, y: 1, z: 5.0, locname: locationname
----------notificationInBytes был равен null
----- получил байты, количество: 4096
----- получил байты, количество: 4096
----- получил байты, количество: 4096
java.io.StreamCorruptedException: invalid stream header: 04000000
at java.io.ObjectInputStream.readStreamHeader(ObjectInputStream.java:899)
at java.io.ObjectInputStream.<init>(ObjectInputStream.java:357)
at com.itmo.utils.SerializationManager2.readObject(SerializationManager2.java:9)
at com.itmo.client.ListenerForNotifications.run(ListenerForNotifications.java:27)
java.io.StreamCorruptedException: invalid stream header: 686A8101
at java.io.ObjectInputStream.readStreamHeader(ObjectInputStream.java:899)
at java.io.ObjectInputStream.<init>(ObjectInputStream.java:357)
at com.itmo.utils.SerializationManager2.readObject(SerializationManager2.java:9)
at com.itmo.client.ListenerForNotifications.run(ListenerForNotifications.java:27)
java.io.StreamCorruptedException: invalid stream header: 72000F6A
at java.io.ObjectInputStream.readStreamHeader(ObjectInputStream.java:899)
at java.io.ObjectInputStream.<init>(ObjectInputStream.java:357)
at com.itmo.utils.SerializationManager2.readObject(SerializationManager2.java:9)
at com.itmo.client.ListenerForNotifications.run(ListenerForNotifications.java:27)
То есть передаются 2 элемента коллекции из 9 нужных.
Вот класс Client (на всякий случай):
/*
Этот класс нужен для того, чтобы подключиться к серверу, иметь возможность послать команду
и получить ее обратно.
Также можно запустить клиента в консольном режиме прям отсюда.
*/
public class Client {
@Getter
private Socket socket;
private static final Scanner scanner = new Scanner(System.in);
public final int BUFFER_SIZE = 4096;
@Getter @Setter
private User user;
private boolean notExit = true;
private final int port;
private boolean once = true;
private final String host;
private MyConsole console;
public Client(String host, int port){
this.host = host;
this.port = port;
user = new User("unregistered", "");
}
public void sendCommandToServer(Command command){
try{
command.setUser(user);
sendCom(command);
} catch (IOException e) {
if (notExit) {
System.out.println("Потеря соединения");
connect();
}
}
}
public void connect() {
System.out.println("Пытаюсь установить соединение с сервером (" + host + ":" + port + ")");
InetAddress addr;
while (true) {
try {
addr = InetAddress.getByName(host);
socket = new Socket(addr, port);
System.out.println("Подключено: " + socket);
once=true;
return;
} catch (UnknownHostException e) {
if(once){
once=false;
System.out.println("Неправильно указан хост");
}
} catch (IOException e) {
if(once){
once=false;
System.out.println("Сервер отключен");
}
}
}
}
private void setUserNameAfterRegistering(String answer){
//todo
if(answer.startsWith("Здравствуйте, ")) {
user.setName(answer.trim().substring(14));
}
}
private void checkExitCommand(Command command){
if(command instanceof ExitCommand) {
notExit = false;
System.exit(0);
}
}
private void sendCom(Command command) throws IOException {
sendOneByte();
sendComBySocket(command);
}
public String getAnswerFromServer(){
try {
return getAns();
} catch (IOException | ClassNotFoundException e) {
System.out.println(e.getMessage());
return null;
}
}
private String getAns() throws IOException, ClassNotFoundException {
byte[] buff = new byte[BUFFER_SIZE];
int got = socket.getInputStream().read(buff);
if(got>0){
Response r = SerializationManager.readObject(buff);
setUserNameAfterRegistering(r.getAnswer()); // это костыль
return r.getAnswer();
}
return null;
}
public void closeEverything(){
scanner.close();
try {
socket.close();
} catch (IOException ignored) {
}
}
private void sendOneByte() throws IOException {
byte[] b = new byte[1];
b[0] = (byte) 127;
socket.getOutputStream().write(b);
}
private void sendComBySocket(Command command) throws IOException {
byte[] serializedCommand = SerializationManager.writeObject(command);
sendBytes(serializedCommand);
}
public void sendBytes(byte[] bytes) throws IOException {
socket.getOutputStream().write(bytes);
}
public byte[] getBytes() throws IOException {
byte[] res = new byte[BUFFER_SIZE];
int got = socket.getInputStream().read(res);
if (got>0) return res;
else return null;
}
}
Вот класс сервера ServerWithThreads для получения клиента и передаче чтения/выполнения команд другому потоку (на всякий случай):
public class ServerWithThreads {
private ServerSocketChannel ssc;
private final Logger log;
private boolean serverOn = true;
private final int port;
public static ExecutorService executorService = Executors.newCachedThreadPool();
public void run(Application application) {
log.info("Запуск сервера, порт: " + port);
setupNet();
SocketChannel clientSocketChannel = null;
while (serverOn) {
try{
clientSocketChannel = accept();
if(clientSocketChannel!=null){
new Thread(new ReadRequestThread(clientSocketChannel, application)).start();
}
}catch (IOException e){
e.printStackTrace();
}
}
closeEverything();
System.exit(0);
}
private void closeEverything() {
try {
if (ssc != null) ssc.close();
} catch (IOException ignored) {
}
ssc = null;
}
public SocketChannel accept() throws IOException {
SocketChannel channel = ssc.accept();
if (channel != null) {
log.info("accepted client");
try {
channel.configureBlocking(false);
channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
return channel;
} catch (IOException e) {
log.error("Unable to accept channel");
}
}
return null;
}
public void setupNet(){
ssc = null;
try {
ssc = ServerSocketChannel.open();
ssc.configureBlocking(false);
InetSocketAddress inetSocketAddress = new InetSocketAddress(port);
ssc.socket().bind(inetSocketAddress);
} catch (IOException e) {
System.out.println("Most likely the server is already running");
}
}
public ServerWithThreads(int port, Logger log){
this.port = port;
this.log = log;
}
}
Класс RequestExecutorThread для выполнения команд, в котором, получая команду подписки на оповещения начинается передача элементов коллекции клиенту.
public class RequestExecutorThread extends Thread {
private final Command command;
private final SocketChannel channel;
private final Application application;
private final User user;
public RequestExecutorThread(Command command, SocketChannel channel, Application application) {
this.command = command;
this.channel = channel;
this.application = application;
user = command.getUser();
}
@Override
public void run() {
if (command != null) {
if(command instanceof SubscribeForNotificationsCommand){
System.out.println("я получил команду Subscribe от " + channel);
application.sendCollectionToClient(channel);
System.out.println("я передал коллекцию по channel " + channel);
application.notificationProducer.subscribeForNotifications(channel);
System.out.println("я выхожу из requestExecutorThread");
return;
}
String res = ServerMain.localeClass.getString("not_registered.text");
boolean userIsRegistered = (!user.getName().equals("unregistered") &&
application.activeUsers.containsUserName(user.getName()));
if(command.isNoRightsToExecute() || userIsRegistered){
res = command.execute(application, user);
}
ServerWithThreads.executorService.execute(new GiveResponseTask(channel, application, user, res));
}
}
}
Класс Application - в нем есть метод sendCollectionToClient(), который вызывается при подписке на оповещения
/*
the purpose of this class is storing collection
and handling active users
*/
public class Application implements Serializable {
@Setter
@Getter
private MyDragonsCollection collection;
private Date date;
public DatabaseManager db;
public ActiveUsersHandler activeUsers;
public NotificationProducer notificationProducer;
public Application() throws SQLException {
db = new DatabaseManager();
collection = new MyDragonsCollection(db.getCollectionFromDatabase());
date = new Date();
activeUsers = ActiveUsersHandler.getInstance();
notificationProducer = new NotificationProducer();
}
public void syncWithDB() {
try {
this.collection = new MyDragonsCollection(db.getCollectionFromDatabase());
} catch (SQLException e) {
System.out.println("ошибка в db");
}
}
public Application(MyDragonsCollection collection) {
this.collection = collection;
}
// after getting SubscribeForNotificationsCommand server sends all collection to socketchannel
public void sendCollectionToClient(SocketChannel channel) {
collection.getDragons().forEach( d -> {
AddNotification addNotification = new AddNotification(d);
notificationProducer.sendAddNotification(channel, addNotification);
});
}
}
Класс ListenerForNotifications, который отправляет запрос на подписку оповещений.
@AllArgsConstructor
public class ListenerForNotifications extends Thread {
String host;
int port;
@Override
public void run() {
Client c = new Client(host, port);
c.connect();
byte[] notificationInBytes;
c.sendCommandToServer(new SubscribeForNotificationsCommand());
SerializationManager2<AddNotification> sm = new SerializationManager2<>();
while(true){
try {
notificationInBytes = c.getBytes();
if (notificationInBytes!=null){
System.out.println("----- получил байты, количество: " + notificationInBytes.length);
Notification notification = sm.readObject(notificationInBytes);
System.out.println("-------вот дракон из этого оповещения: \n" + notification.getDragon());
notification.updateData();
}
System.out.println("----------notificationInBytes был равен null");
} catch (IOException | ClassNotFoundException e) {
e.printStackTrace();
}
}
}
}
Этот поток запускается в главном классе JavaFX.
public class UIApp extends Application {
/*
много кода для настройки GUI
*/
@Override
public void start(Stage primaryStage) throws Exception {
resourceBundle = ResourceBundle.getBundle("locals", Locale.forLanguageTag("SE"), new UTF8Control());
mainStage = primaryStage;
authorizationStage = WindowsCreator.createAuthorization();
authorizationStage.show();
initPrimaryStage(primaryStage);
ListenerForNotifications notificationListener = new ListenerForNotifications(MainUI.host, MainUI.port);
notificationListener.setDaemon(true);
notificationListener.start();
}
Класс для оповещений AddNotification:
public class AddNotification extends Notification{
public AddNotification(Dragon dragon) {
super(dragon);
}
@Override
public void updateData() {
//todo
UIApp.mainWindowController.addDragonToColl(dragon);
}
}
@AllArgsConstructor
public abstract class Notification implements Serializable {
public abstract void updateData();
@Getter
protected Dragon dragon;
}
Класс для отправки оповещений NotificationProducer:
public class NotificationProducer implements Serializable {
public Set<SocketChannel> subscribers;
public NotificationProducer() {
subscribers = new HashSet<>();
}
public void sendRemoveNotificationToAll(RemoveNotification removeNotification){
subscribers.forEach( sub -> {
sendRemoveNotification(sub, removeNotification);
});
}
public void sendAddNotificationToAll(AddNotification addNotification){
subscribers.forEach( sub -> {
sendAddNotification(sub, addNotification);
});
}
public void sendAddNotification(SocketChannel socketChannel, AddNotification addNotification){
try {
SerializationManager2<AddNotification> sm = new SerializationManager2<>();
ByteBuffer byteBuffer = ByteBuffer.wrap(sm.writeObject(addNotification));
socketChannel.write(byteBuffer);
} catch (IOException e) {
e.printStackTrace();
}
}
public void sendRemoveNotification(SocketChannel socketChannel, RemoveNotification removeNotification){
try {
ByteBuffer byteBuffer = ByteBuffer.wrap(SerializationManager.writeObject(removeNotification));
socketChannel.write(byteBuffer);
} catch (IOException e) {
e.printStackTrace();
}
}
public void subscribeForNotifications(SocketChannel socketChannel){
subscribers.add(socketChannel);
System.out.println("я зарегистрировал подписчика " + socketChannel);
}
}
Класс, который использую для сериализации и десериализации команд и оповещений:
public class SerializationManager2<T> {
public T readObject(byte[] data) throws IOException, ClassNotFoundException, ClassCastException {
ByteArrayInputStream byteStream = new ByteArrayInputStream(data);
ObjectInputStream obs = new ObjectInputStream(byteStream);
return (T) obs.readObject();
}
public byte[] writeObject(T object) throws IOException {
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
ObjectOutputStream objectOutputStream = new ObjectOutputStream(outputStream);
objectOutputStream.writeObject(object);
return outputStream.toByteArray();
}
}
Я понимаю, что ошибка возникает из-за того, что в заголовках не оказывается нужного значения, свидетельствующего о том, что передаются сериализованные объекты. Но как это исправить я не могу понять. Подозреваю, что это происходит из-за non-blocking режима в SocketChannel. Спасибо что прочитали. UPD: при дебаге тоже бывают ошибки. "notificationInBytes был равен null" выводит всегда, забыл else.