package envoy.client.net; import java.io.*; import java.net.Socket; import java.util.concurrent.TimeoutException; import java.util.logging.*; import envoy.client.data.*; import envoy.client.event.*; import envoy.data.*; import envoy.event.*; import envoy.event.Event; import envoy.event.contact.*; import envoy.util.*; import dev.kske.eventbus.*; /** * Establishes a connection to the server, performs a handshake and delivers * certain objects to the server. *

* Project: envoy-client
* File: Client.java
* Created: 28 Sep 2019
* * @author Kai S. K. Engelbart * @author Maximilian Käfer * @author Leon Hofmeister * @since Envoy Client v0.1-alpha */ public final class Client implements EventListener, Closeable { // Connection handling private Socket socket; private Receiver receiver; private boolean online; // Asynchronously initialized during handshake private volatile User sender; private volatile boolean rejected; // Configuration, logging and event management private static final ClientConfig config = ClientConfig.getInstance(); private static final Logger logger = EnvoyLog.getLogger(Client.class); private static final EventBus eventBus = EventBus.getInstance(); /** * Constructs a client and registers it as an event listener. * * @since Envoy Client v0.2-beta */ public Client() { eventBus.registerListener(this); } /** * Enters the online mode by acquiring a user ID from the server. As a * connection has to be established and a handshake has to be made, this method * will block for up to 5 seconds. If the handshake does exceed this time limit, * an exception is thrown. * * @param credentials the login credentials of the user * @param cacheMap the map of all caches needed * @throws TimeoutException if the server could not be reached * @throws IOException if the login credentials could not be written * @throws InterruptedException if the current thread is interrupted while * waiting for the handshake response */ public void performHandshake(LoginCredentials credentials, CacheMap cacheMap) throws TimeoutException, IOException, InterruptedException { if (online) throw new IllegalStateException("Handshake has already been performed successfully"); // Establish TCP connection logger.log(Level.FINER, String.format("Attempting connection to server %s:%d...", config.getServer(), config.getPort())); socket = new Socket(config.getServer(), config.getPort()); logger.log(Level.FINE, "Successfully established TCP connection to server"); // Create object receiver receiver = new Receiver(socket.getInputStream()); // Register user creation processor, contact list processor, message cache and // authentication token receiver.registerProcessor(User.class, sender -> this.sender = sender); receiver.registerProcessors(cacheMap.getMap()); receiver.registerProcessor(HandshakeRejection.class, evt -> { rejected = true; eventBus.dispatch(evt); }); receiver.registerProcessor(NewAuthToken.class, eventBus::dispatch); rejected = false; // Start receiver receiver.start(); // Write login credentials SerializationUtils.writeBytesWithLength(credentials, socket.getOutputStream()); // Wait for a maximum of five seconds to acquire the sender object final long start = System.currentTimeMillis(); while (sender == null) { // Quit immediately after handshake rejection // This method can then be called again if (rejected) { socket.close(); receiver.removeAllProcessors(); return; } if (System.currentTimeMillis() - start > 5000) throw new TimeoutException("Did not log in after 5 seconds"); Thread.sleep(500); } online = true; logger.log(Level.INFO, "Handshake completed."); } /** * Initializes the {@link Receiver} used to process data sent from the server to * this client. * * @param localDB the local database used to persist the current * {@link IDGenerator} * @param cacheMap the map of all caches needed * @throws IOException if no {@link IDGenerator} is present and none could be * requested from the server * @since Envoy Client v0.2-alpha */ public void initReceiver(LocalDB localDB, CacheMap cacheMap) throws IOException { checkOnline(); // Remove all processors as they are only used during the handshake receiver.removeAllProcessors(); // Process incoming messages final var receivedMessageProcessor = new ReceivedMessageProcessor(); final var receivedGroupMessageProcessor = new ReceivedGroupMessageProcessor(); final var messageStatusChangeProcessor = new MessageStatusChangeProcessor(); final var groupMessageStatusChangeProcessor = new GroupMessageStatusChangeProcessor(); receiver.registerProcessor(GroupMessage.class, receivedGroupMessageProcessor); receiver.registerProcessor(Message.class, receivedMessageProcessor); receiver.registerProcessor(MessageStatusChange.class, messageStatusChangeProcessor); receiver.registerProcessor(GroupMessageStatusChange.class, groupMessageStatusChangeProcessor); // Relay cached messages and message status changes cacheMap.get(Message.class).setProcessor(receivedMessageProcessor); cacheMap.get(GroupMessage.class).setProcessor(receivedGroupMessageProcessor); cacheMap.get(MessageStatusChange.class).setProcessor(messageStatusChangeProcessor); cacheMap.get(GroupMessageStatusChange.class).setProcessor(groupMessageStatusChangeProcessor); // Process user status changes receiver.registerProcessor(UserStatusChange.class, eventBus::dispatch); // Process message ID generation receiver.registerProcessor(IDGenerator.class, localDB::setIDGenerator); // Process name changes receiver.registerProcessor(NameChange.class, evt -> { localDB.replaceContactName(evt); eventBus.dispatch(evt); }); // Process contact searches receiver.registerProcessor(UserSearchResult.class, eventBus::dispatch); // Process contact operations receiver.registerProcessor(ContactOperation.class, eventBus::dispatch); // Process group size changes receiver.registerProcessor(GroupResize.class, evt -> { localDB.updateGroup(evt); eventBus.dispatch(evt); }); // Process IsTyping events receiver.registerProcessor(IsTyping.class, eventBus::dispatch); // Process PasswordChangeResults receiver.registerProcessor(PasswordChangeResult.class, eventBus::dispatch); // Process ProfilePicChanges receiver.registerProcessor(ProfilePicChange.class, eventBus::dispatch); // Process requests to not send any more attachments as they will not be shown // to other users receiver.registerProcessor(NoAttachments.class, eventBus::dispatch); // Process group creation results - they might have been disabled on the server receiver.registerProcessor(GroupCreationResult.class, eventBus::dispatch); // Request a generator if none is present or the existing one is consumed if (!localDB.hasIDGenerator() || !localDB.getIDGenerator().hasNext()) requestIdGenerator(); // Relay caches cacheMap.getMap().values().forEach(Cache::relay); } /** * Sends a message to the server. The message's status will be incremented once * it was delivered successfully. * * @param message the message to send * @throws IOException if the message does not reach the server * @since Envoy Client v0.3-alpha */ public void sendMessage(Message message) throws IOException { writeObject(message); message.nextStatus(); } /** * Sends an event to the server. * * @param evt the event to send * @throws IOException if the event did not reach the server */ public void sendEvent(Event evt) throws IOException { if (online) writeObject(evt); } /** * Requests a new {@link IDGenerator} from the server. * * @throws IOException if the request does not reach the server * @since Envoy Client v0.3-alpha */ public void requestIdGenerator() throws IOException { logger.log(Level.INFO, "Requesting new id generator..."); writeObject(new IDGeneratorRequest()); } /** * Sends the value of a send event to the server. * * @param evt the send event to extract the value from * @since Envoy Client v0.2-beta */ @dev.kske.eventbus.Event private void onSendEvent(SendEvent evt) { try { sendEvent(evt.get()); } catch (final IOException e) { logger.log(Level.WARNING, "An error occurred when trying to send " + evt, e); } } @Override @dev.kske.eventbus.Event(eventType = EnvoyCloseEvent.class, priority = 800) public void close() { if (online) { logger.log(Level.INFO, "Closing connection..."); try { socket.close(); } catch (final IOException e) {} } } private void writeObject(Object obj) throws IOException { checkOnline(); logger.log(Level.FINE, "Sending " + obj); SerializationUtils.writeBytesWithLength(obj, socket.getOutputStream()); } private void checkOnline() { if (!online) throw new IllegalStateException("Client is not online"); } /** * @return the {@link User} as which this client is logged in * @since Envoy Client v0.1-alpha */ public User getSender() { return sender; } /** * Sets the client user which is used to send messages. * * @param clientUser the client user to set * @since Envoy Client v0.2-alpha */ public void setSender(User clientUser) { sender = clientUser; } /** * @return the {@link Receiver} used by this {@link Client} */ public Receiver getReceiver() { return receiver; } /** * @return {@code true} if a connection to the server could be established * @since Envoy Client v0.2-alpha */ public boolean isOnline() { return online; } }