connection.java

MegaMek is a networked Java clone of BattleTech, a turn-based sci-fi boardgame for 2+ players. Fight

    
    /**
     * Removes the specified connection listener.
     *
     * @param listener the connection listener.
     */
    public void removeConnectionListener(ConnectionListener listener) {
        connectionListeners.removeElement(listener);
    }
    
    /**
     * Reports receive exception to the <code>System.err</code>
     * @param ex <code>Exception</code>
     * @param packet <code>Packet</code>
     */
    protected void reportSendException(Exception ex, SendPacket packet) {
        System.err.print(getConnectionTypeAbbrevation());
        System.err.print(" error sending command #");
        System.err.print(packet.getCommand()); //$NON-NLS-1$
        System.err.print(": ");
        System.err.println(ex.getMessage()); //$NON-NLS-1$
        reportLastCommands();
    }

    /**
     * Reports receive exception to the <code>System.err</code> 
     * @param ex <code>Exception</code>
     */
    protected void reportReceiveException(Exception ex) {
        StringBuffer message = new StringBuffer();
        reportReceiveException(ex, message);
        System.err.println(message);
    }

    /**
     * Appends the receive exception report to the given <code>StringBuffer</code> 
     * @param ex <code>Exception</code>
     */
    protected void reportReceiveException(Exception ex, StringBuffer buffer) {
        System.err.print(getConnectionTypeAbbrevation());
        System.err.print(" error reading command: ");
        System.err.println(ex.getMessage()); //$NON-NLS-1$
        reportLastCommands();
    }

    /**
     * Appends the last commands sent/received to the given <code>StringBuffer</code>
     * @param buffer <code>StringBuffer</code> to add the report to 
     */
    protected void reportLastCommands() {
        reportLastCommands(true);
        System.err.println();
        reportLastCommands(false);
        System.err.println();
        sendQueue.reportContents();
    }

    /**
     * Appends the last commands sent or received to the given <code>StringBuffer</code>
     * dependig on the <code>sent</code> parameter
     * @param buffer <code>StringBuffer</code> to add the report to
     * @param sent indicates which commands (sent/received) should be reported
     */
    protected void reportLastCommands(boolean sent) {
        CircularIntegerBuffer buf = sent?debugLastFewCommandsSent:debugLastFewCommandsReceived;
        System.err.print("    Last "); //$NON-NLS-1$
        System.err.print(buf.length());
        System.err.print(" commands that were "); //$NON-NLS-1$ 
        System.err.print(sent?"sent":"received"); //$NON-NLS-1$
        System.err.print(" (oldest first): "); //$NON-NLS-1$
        System.err.println(buf); 
    }

    /**
     * Returns the the connection type abbrevation (client/server) that used
     * in the debug messages and so on. 
     * @return
     */
    protected String getConnectionTypeAbbrevation() {
        return isServer()?"s:":"c:"; //$NON-NLS-1$ //$NON-NLS-2$
    }

    /**
     * Returns an input stream
     * @return an input stream
     * @throws IOException
     */
    protected InputStream getInputStream() throws IOException {
        return socket.getInputStream();
    }

    /**
     * Returns an output stream
     * @return an output stream
     * @throws IOException
     */
    protected OutputStream getOutputStream() throws IOException {
        return socket.getOutputStream();
    }

    /**
     * Initializes the sender and receiver threads
     */
    private void initThreads() {

        Runnable receiverRunnable = new Runnable() {
            public void run() {
                while (receiver == Thread.currentThread()) {
                    INetworkPacket np=null;
                    try {
                        np = readNetworkPacket();
                        if (np != null) {
                            processPacket(np);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        reportReceiveException(e);
                        close();
                    }                   
                }
            }

            protected void processPacket(INetworkPacket np) throws Exception {
                PacketMarshaller pm = marshallerFactory.getMarshaller(np.getMarshallingType());
                megamek.debug.Assert.assertTrue(pm != null, "Unknown marshalling type");
                Packet packet = null;
                byte[] data = np.getData();
                bytesReceived += data.length;
                ByteArrayInputStream bis = new ByteArrayInputStream(data);
                InputStream in;
                if (np.isCompressed()) {
                    in = new GZIPInputStream(bis);                                
                } else {
                    in = bis;
                }
                packet = pm.unmarshall(in);
                if (packet != null) {
                    debugLastFewCommandsReceived.push(packet.getCommand());
                    processConnectionEvent(new PacketReceivedEvent(Connection.this, packet));
                }
                
            }
        };        
        receiver = new Thread(receiverRunnable, "Packet Receiver (" + getId() + ")"); //$NON-NLS-1$ //$NON-NLS-2$

        Runnable senderRunable = new Runnable() {
            public void run() {
                while (sender == Thread.currentThread()) {
                    SendPacket packet = sendQueue.getPacket();
                    if (packet != null) {
                        try {
                            processPacket(packet);
                        }catch (Exception e) {
                            reportSendException(e, packet);
                            close();
                        }
                    }
                }
            }

            protected void processPacket(SendPacket packet) throws Exception {
                sendNow(packet);
            }
        };        
        sender = new Thread(senderRunable, "Packet Sender (" + getId() + ")"); //$NON-NLS-1$ //$NON-NLS-2$
        
        sender.start();
        receiver.start();
    }
    
    /**
     * Reads a complete <code>NetworkPacket</code>
     * @return  the <code>NetworkPacket</code> that was sent.
     */
    protected abstract INetworkPacket readNetworkPacket() throws Exception;

    /**
     * Sends the data
     * @param data data to send
     * @param zipped should the data be compressed
     * @throws Exception
     */
    protected abstract void sendNetworkPacket(byte[] data, boolean zipped) throws Exception;
    
    private static class SendQueue {

        private Vector<SendPacket> queue = new Vector();
        private boolean finished = false;

        public synchronized void addPacket(SendPacket packet) {
            queue.addElement(packet);
            notifyAll();
        }

        public synchronized void finish() {
            queue.removeAllElements();
            finished = true; 
            notifyAll();
        }

        /**
         * Waits for a packet to appear in the queue and then returns it.
         * @return the first available packet in the queue
         */
        public synchronized SendPacket getPacket() {
            SendPacket packet = null;
            while (!hasPending() && !finished) {
                try {
                    wait();
                } catch (InterruptedException ex) {
                }
            }
            if (!finished) {
                packet = queue.elementAt(0);
                queue.removeElementAt(0);
            }
            return packet;
        }
        /**
         * Returns true if this connection has pending data
         */
        public synchronized boolean hasPending() {
            return queue.size() > 0;
        }
        
        public void reportContents() {
            System.err.print("Contents of Send Queue: ");
            for(SendPacket p:queue) {
                System.err.print(p.command);
            }
            System.err.println();
        }
    }

    /**
     * Processes game events occurring on this
     * connection by dispatching them to any registered
     * GameListener objects.
     *
     * @param event the game event.
     */
    protected void processConnectionEvent(ConnectionEvent event) {
        for (Enumeration e = connectionListeners.elements(); e.hasMoreElements();) {
            ConnectionListener l = (ConnectionListener) e.nextElement();
            switch (event.getType()) {
            case ConnectionEvent.CONNECTED:
                l.connected((ConnectedEvent)event);
                break;
            case ConnectionEvent.DISCONNECTED:
                l.disconnected((DisconnectedEvent)event);
                break;
            case ConnectionEvent.PACKET_RECEIVED:
                l.packetReceived((PacketReceivedEvent)event);
                break;                
            }
        }
    }
    
    private class SendPacket implements INetworkPacket {
        byte[] data;
        boolean zipped = false;
        int command;
        public SendPacket(Packet packet) {
            command = packet.getCommand();
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            OutputStream out;
            try {
                if (zipData && packet.getData() != null) {
                    out = new GZIPOutputStream(bos);
                    zipped = true;
                } else {
                    out = bos;                            
                }
                marshaller.marshall(packet, out);
                out.close();
                data = bos.toByteArray();
                bytesSent += data.length;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        public int getMarshallingType() {
            return marshallingType;
        }
        public byte[] getData() {
            return data;
        }
        public boolean isCompressed() {
            return zipped;
        }
        public int getCommand() {
            return command;
        }
    }

    /**
     * Connection layer data packet. 
     */
    protected interface INetworkPacket {

        /**
         * Returns data marshalling type
         * @return data marshalling type
         */
        public abstract int getMarshallingType();

        /**
         * Returns packet data
         * @return packet data
         */
        public abstract byte[] getData();

        /**
         * Returns <code>true</code> if data is compressed
         * @return <code>true</code> if data is compressed
         */
        public abstract boolean isCompressed();
    };
}