synchronization

5 files changed, 78 insertions, 71 deletions

diff --git a/src/main/java/com/github/jodersky/flow/low/NativeSerial.java b/src/main/java/com/github/jodersky/flow/low/NativeSerial.java
index b488d5e..132c0ef 100644
--- a/src/main/java/com/github/jodersky/flow/low/NativeSerial.java
+++ b/src/main/java/com/github/jodersky/flow/low/NativeSerial.java
@@ -53,8 +53,10 @@ class NativeSerial {
 	 * this function, the 'serial' pointer will become invalid, make sure you only call it once. This function is NOT
 	 * thread safe, make sure no read or write is in prgress when this function is called (the reason is that per 
 	 * close manual page, close should not be called on a file descriptor that is in use by another thread). 
-	 * @param serial pointer to serial configuration that is to be closed (and freed) */
-	native static void close(long serial);
+	 * @param serial pointer to serial configuration that is to be closed (and freed)
+	 * @return 0 on success
+	 * @return E_IO on error */
+	native static int close(long serial);
 	
 	/**Sets debugging option. If debugging is enabled, detailed error message are printed from method calls. */
 	native static void debug(boolean value);
diff --git a/src/main/native/flow.c b/src/main/native/flow.c
index 4353e80..43dd750 100644
--- a/src/main/native/flow.c
+++ b/src/main/native/flow.c
@@ -143,19 +143,23 @@ int serial_open(const char* port_name, int baud, struct serial_config** serial)
   return 0;
 }
 
-void serial_close(struct serial_config* serial) {
+int serial_close(struct serial_config* serial) {
   if (close(serial->pipe_write_fd) < 0) {
     DEBUG(perror("close write end of pipe"););
+    return E_IO;
   }
   if (close(serial->pipe_read_fd) < 0) {
     DEBUG(perror("close read end of pipe"););
+    return E_IO;
   }
   
   if (flock(serial->port_fd, LOCK_UN) < 0){
     DEBUG(perror("release lock on port"););
+    return E_IO;
   }
   if (close(serial->port_fd) < 0) {
     DEBUG(perror("close port"););
+    return E_IO;
   }
   
   free(serial);
@@ -194,6 +198,15 @@ int serial_read(struct serial_config* serial, unsigned char* buffer, size_t size
   }
 }
 
+int serial_write(struct serial_config* serial, unsigned char* data, size_t size) {
+  int r = write(serial->port_fd, data, size);
+  if (r < 0) {
+    DEBUG(perror("write"););
+    return E_IO;
+  }
+  return r;
+}
+
 int serial_interrupt(struct serial_config* serial) {
   int data = 0xffffffff;
   
@@ -206,16 +219,6 @@ int serial_interrupt(struct serial_config* serial) {
   return 0;
 }
 
-int serial_write(struct serial_config* serial, unsigned char* data, size_t size) {
-  int r = write(serial->port_fd, data, size);
-  if (r < 0) {
-    DEBUG(perror("write"););
-    return E_IO;
-  }
-  return r;
-}
-
-
 
 // JNI bindings
 // ============
@@ -242,7 +245,7 @@ JNIEXPORT jint JNICALL Java_com_github_jodersky_flow_low_NativeSerial_open
   return r;
 }
 
-JNIEXPORT void JNICALL Java_com_github_jodersky_flow_low_NativeSerial_close
+JNIEXPORT jint JNICALL Java_com_github_jodersky_flow_low_NativeSerial_close
   (JNIEnv * env, jclass clazz, jlong serial)
 {
   serial_close(j2s(serial));
@@ -276,6 +279,12 @@ JNIEXPORT jint JNICALL Java_com_github_jodersky_flow_low_NativeSerial_write
   return r;
 }
 
+JNIEXPORT jint JNICALL Java_com_github_jodersky_flow_low_NativeSerial_interrupt
+  (JNIEnv * env, jclass clazz, jlong serial)
+{
+  return serial_interrupt(j2s(serial));
+}
+
 JNIEXPORT void JNICALL Java_com_github_jodersky_flow_low_NativeSerial_debug
   (JNIEnv *env, jclass clazz, jboolean value)
 {
diff --git a/src/main/native/flow.h b/src/main/native/flow.h
index 97eae83..d1823c1 100644
--- a/src/main/native/flow.h
+++ b/src/main/native/flow.h
@@ -38,8 +38,10 @@ int serial_open(const char* port_name, int baud, struct serial_config** serial);
  * this function, the 'serial' pointer will become invalid, make sure you only call it once. This function is NOT
  * thread safe, make sure no read or write is in prgress when this function is called (the reason is that per 
  * close manual page, close should not be called on a file descriptor that is in use by another thread). 
- * @param serial pointer to serial configuration that is to be closed (and freed) */
-void serial_close(struct serial_config* serial);
+ * @param serial pointer to serial configuration that is to be closed (and freed)
+ * @return 0 on success
+ * @return E_IO on error */
+int serial_close(struct serial_config* serial);
 
 /**Starts a blocking read from a previously opened serial port. The read is blocking, however it may be
  * interrupted by calling 'serial_interrupt' on the given serial port.
diff --git a/src/main/scala/com/github/jodersky/flow/exceptions.scala b/src/main/scala/com/github/jodersky/flow/exceptions.scala
index ab9392f..ebcce60 100644
--- a/src/main/scala/com/github/jodersky/flow/exceptions.scala
+++ b/src/main/scala/com/github/jodersky/flow/exceptions.scala
@@ -2,8 +2,9 @@ package com.github.jodersky.flow
 
 import java.io.IOException
 
-class NoSuchPortException(message: String) extends IOException(message)
-class PortInUseException(message: String) extends IOException(message)
-class AccessDeniedException(message: String) extends IOException(message)
-class IllegalBaudRateException(message: String) extends IOException(message)
-class PortClosingException(message: String) extends IOException(message)
\ No newline at end of file
+//class NoSuchPortException(message: String) extends IOException(message)
+class PortInUseException(message: String) extends Exception(message)
+class AccessDeniedException(message: String) extends Exception(message)
+class IllegalBaudRateException(message: String) extends Exception(message)
+class PortInterruptedException(message: String) extends Exception(message)
+class PortClosedException(message: String) extends Exception(message)
\ No newline at end of file
diff --git a/src/main/scala/com/github/jodersky/flow/low/Serial.scala b/src/main/scala/com/github/jodersky/flow/low/Serial.scala
index 5a490ca..916ab84 100644
--- a/src/main/scala/com/github/jodersky/flow/low/Serial.scala
+++ b/src/main/scala/com/github/jodersky/flow/low/Serial.scala
@@ -4,51 +4,49 @@ import scala.concurrent._
 import scala.concurrent.ExecutionContext.Implicits._
 import java.io.IOException
 import com.github.jodersky.flow.AccessDeniedException
-import com.github.jodersky.flow.NoSuchPortException
 import com.github.jodersky.flow.PortInUseException
-import com.github.jodersky.flow.PortClosingException
+import com.github.jodersky.flow.PortClosedException
 import com.github.jodersky.flow.IllegalBaudRateException
 import scala.util.Try
-
+import java.util.concurrent.atomic.AtomicBoolean
+import com.github.jodersky.flow.PortInterruptedException
 
 class Serial private (val port: String, private val pointer: Long) {
-  import NativeSerial._
-  
-  /** State of current connection, required so that close may not be called multiple
-   *  times and used to ensure close and read are never called at the same moment. */
-  private var _closed = false;
-  private def closed = synchronized{_closed}
-  private def closed_=(newValue: Boolean) = synchronized{_closed = newValue} 
-  
+  import Serial._
 
-  def read(): Array[Byte] = {
-    if (!closed) {
-      //read
-    } else {
-      //
-    }
-    
-    val buffer = new Array[Byte](100)
-    
-    NativeSerial.read(pointer, buffer) match {
-      case E_INTERRUPT => throw new PortClosingException(port)
-      case bytes if bytes > 0 => buffer.take(bytes)
-      case error => throw new IOException(s"unknown error code: ${error}")
-    }
+  private val reading = new AtomicBoolean(false)
+  private val writing = new AtomicBoolean(false)
+  private val closed = new AtomicBoolean(false)
+
+  def close(): Unit = if (!closed.get()) {
+    closed.set(true)
+    requireSuccess(NativeSerial.interrupt(pointer), port)
+    if (writing.get()) wait() // if reading, wait for read to finish
+    if (reading.get()) wait()
+    requireSuccess(NativeSerial.close(pointer), port)
   }
 
-  def write(data: Array[Byte]): Array[Byte] = {
-    //
-    import NativeSerial._
-    NativeSerial.write(pointer, data) match {
-      case E_IO => throw new IOException(port)
-      case bytes if bytes > 0 => data.take(bytes)
-      case error => throw new IOException(s"unknown write error ${error}")
-    }
+  def read(): Array[Byte] = if (!closed.get) {
+    reading.set(true)
+    val buffer = new Array[Byte](100)
+    val readResult = NativeSerial.read(pointer, buffer)
+    if (closed.get) notify(); //read was interrupted by close
+    reading.set(false)
+    val n = requireSuccess(readResult, port)
+    buffer take n
+  } else {
+    throw new PortClosedException(s"port ${port} is already closed")
   }
 
-  def close() = {
-    NativeSerial.close(pointer)
+  def write(data: Array[Byte]): Array[Byte] = if (!closed.get) {
+    writing.set(true)
+    val writeResult = NativeSerial.write(pointer, data)
+    if (closed.get) notify()
+    writing.set(false)
+    val n = requireSuccess(writeResult, port)
+    data take n
+  } else {
+    throw new PortClosedException(s"port ${port} is already closed")
   }
 
 }
@@ -56,25 +54,20 @@ class Serial private (val port: String, private val pointer: Long) {
 object Serial {
   import NativeSerial._
 
-  def except(result: Int): Int = result match {
-    case E_IO => throw new IOException("")
-    case E_ACCESS_DENIED => 0
-    case E_BUSY => 0
-    case E_INVALID_BAUD => 0
-    case E_INTERRUPT => 0
+  def requireSuccess(result: Int, port: String): Int = result match {
+    case E_IO => throw new IOException(port)
+    case E_ACCESS_DENIED => throw new AccessDeniedException(port)
+    case E_BUSY => throw new PortInUseException(port)
+    case E_INVALID_BAUD => throw new IllegalBaudRateException("use standard baud rate (see termios.h)")
+    case E_INTERRUPT => throw new PortInterruptedException(port)
+    case error if error < 0 => throw new IOException(s"unknown error code: ${error}")
+    case success => success
   }
 
   def open(port: String, baud: Int): Serial = synchronized {
     val pointer = new Array[Long](1)
-    NativeSerial.open(port, baud, pointer) match {
-      case E_IO => throw new IOException("")
-      case E_ACCESS_DENIED => throw new AccessDeniedException(port)
-      case E_BUSY => throw new PortInUseException(port)
-      case E_INVALID_BAUD => throw new IllegalBaudRateException(baud.toString)
-      //handle no such port
-      case 0 => new Serial(port, pointer(0))
-      case error => throw new IOException(s"unknown error code: ${error}")
-    }
+    requireSuccess(NativeSerial.open(port, baud, pointer), port)
+    new Serial(port, pointer(0))
   }
 
   def debug(value: Boolean) = NativeSerial.debug(value)