src/uk/ac/ntu/n0521366/wsyd/libs/net/NetworkStream.java

   1 /*
   2  * The MIT License
   3  *
   4  * Copyright 2015 eddie.
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining a copy
   7  * of this software and associated documentation files (the "Software"), to deal
   8  * in the Software without restriction, including without limitation the rights
   9  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  10  * copies of the Software, and to permit persons to whom the Software is
  11  * furnished to do so, subject to the following conditions:
  12  *
  13  * The above copyright notice and this permission notice shall be included in
  14  * all copies or substantial portions of the Software.
  15  *
  16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  17  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  18  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  19  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  20  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  21  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  22  * THE SOFTWARE.
  23  */
  24 package uk.ac.ntu.n0521366.wsyd.libs.net;
  25
  26 import java.io.ByteArrayOutputStream;
  27 import java.io.IOException;
  28 import java.io.InputStream;
  29 import java.io.ObjectInputStream;
  30 import java.io.ObjectOutputStream;
  31 import java.io.OutputStream;
  32 import java.net.Socket;
  33 import java.util.concurrent.ConcurrentLinkedQueue;
  34 import java.util.logging.Level;
  35 import java.util.logging.Logger;
  36 import javax.swing.SwingWorker;
  37
  38 /**
  39  *
  40  * @author eddie
  41  */
  42 public class NetworkStream extends SwingWorker<Boolean, NetworkMessage> {
  43
  44     private final Socket _socket;
  45
  46     static final long longSize = Long.SIZE / 8;
  47
  48     int _key;
  49
  50     /**
  51      * Thread safe First In, First Out Queue of NetworkMessage objects waiting to be sent.
  52      *
  53      * Allows the Owner Thread to submit new messages for sending that the Worker Thread
  54      * can safely access.
  55      */
  56     protected ConcurrentLinkedQueue<NetworkMessage> _sendMessageQueue = new ConcurrentLinkedQueue<>();
  57
  58     private ObjectOutputStream _socketOS;
  59
  60     private InputStream _socketIS;
  61
  62     private final NetworkSocketClosing _owner;
  63
  64     private ObjectInputStream _ois;
  65
  66     private ObjectOutputStream _oos;
  67
  68     private ByteArrayOutputStream _baos;
  69
  70     NetworkStream(Socket socket, NetworkSocketClosing owner) {
  71         this._socket = socket;
  72         this._owner = owner;
  73     }
  74
  75     @Override
  76     public Boolean doInBackground() throws Exception {
  77
  78         _baos = new ByteArrayOutputStream();
  79         _oos = new ObjectOutputStream(_baos);
  80         _socketOS = new ObjectOutputStream(_socket.getOutputStream());
  81
  82         _socketIS = _socket.getInputStream();
  83         _ois = new ObjectInputStream(_socketIS);
  84
  85         long expectedLength = longSize;
  86         boolean expectingLong = true;
  87
  88         while (!this.isCancelled()) {
  89
  90             // check for incoming message
  91             if (_socketIS.available() >= expectedLength) {
  92                 if (expectingLong == true) {
  93                     expectedLength = _ois.readLong();
  94                     expectingLong = false;
  95                 }
  96                 else {
  97                     read();
  98                     expectedLength = longSize;
  99                     expectingLong = true;
 100                 }
 101             }
 102             // send a queued message
 103             NetworkMessage temp = this.dequeueMessage();
 104             if (temp != null) {
 105                 if (!this.write(temp)) {
 106                     System.err.println("Unable to send message over TCP channel");
 107                     // FIXME: Potentially add logger support
 108                 }
 109             }
 110
 111         }
 112         return false;
 113     }
 114
 115     @Override
 116     protected void done() {
 117         _owner.canCloseSocket(_key);
 118     }
 119
 120     /**
 121      * Send a NetworkMessage over a NetworkStream.
 122      *
 123      * Each message is preceeded by its length as a long to avoid rare, but possible blocking
 124      * in the reader.
 125      *
 126      * @param message
 127      * @return true if message successfully sent
 128      */
 129     public boolean write(NetworkMessage message) {
 130         boolean result = false;
 131
 132         if (message != null) {
 133             try {
 134                 _oos.writeObject(message);
 135                 _oos.flush();
 136                 long messageSize = _baos.size();
 137                 _socketOS.writeLong(messageSize);
 138                 _baos.writeTo(_socketOS);
 139
 140                 result = true;
 141
 142             } catch (IOException ex) {
 143                 // TODO: Implement graceful disconnect
 144             }
 145         }
 146         return result;
 147     }
 148
 149     /**
 150      * Reads NetworkMessage from the incoming stream and publishes it.
 151      *
 152      *
 153      * @return true if message successfully published
 154      */
 155     public boolean read() {
 156         boolean result = false;
 157
 158         try {
 159             NetworkMessage message = (NetworkMessage)_ois.readObject();
 160             publish(message);
 161
 162             result = true;
 163         } catch (IOException ex) {
 164             Logger.getLogger(NetworkStream.class.getName()).log(Level.SEVERE, null, ex);
 165             // FIXME: Replace logger
 166             // TODO: Graceful disconnect
 167         } catch (ClassNotFoundException ex) {
 168             Logger.getLogger(NetworkStream.class.getName()).log(Level.SEVERE, null, ex);
 169             // FIXME: Replace logger
 170         }
 171         // TODO: Implement NetworkStream::read()
 172         return result;
 173     }
 174
 175     /**
 176      * Removes a message from the queue of pending messages.
 177      *
 178      * This method is called on the Worker Thread by the doInBackground() main loop.
 179      *
 180      * @return a message to be sent
 181      */
 182     protected NetworkMessage dequeueMessage() {
 183         return this._sendMessageQueue.poll();
 184     }
 185
 186     /* XXX: Methods below here all execute on the GUI Event Dispatch Thread */
 187
 188     /**
 189      * Adds a message to the queue of pending messages.
 190      *
 191      * This method will usually be called from the Owner Thread.
 192      *
 193      * @param message to be sent
 194      * @return true if the message was added to the queue
 195      */
 196     public boolean queueMessage(NetworkMessage message) throws IllegalArgumentException {
 197         boolean result = false;
 198
 199         if (message != null) {
 200             NetworkMessage temp;
 201             try { // make a deep clone of the message
 202                 temp = NetworkMessage.clone(message);
 203                 result = this._sendMessageQueue.add(temp);
 204             } catch (CloneNotSupportedException e) {
 205                 // TODO: queueMessage() log CloneNotSupportedException
 206                 e.printStackTrace();
 207             }
 208         }
 209         return result;
 210     }
 211
 212 }