diff --git a/usbSerialForAndroid/build.gradle b/usbSerialForAndroid/build.gradle index 049e55e6..06ad81f9 100644 --- a/usbSerialForAndroid/build.gradle +++ b/usbSerialForAndroid/build.gradle @@ -34,9 +34,10 @@ dependencies { implementation "androidx.annotation:annotation:1.8.0" testImplementation 'junit:junit:4.13.2' testImplementation 'org.mockito:mockito-core:5.12.0' + androidTestImplementation 'androidx.appcompat:appcompat:1.6.1' androidTestImplementation 'androidx.test:core:1.5.0' androidTestImplementation 'androidx.test:runner:1.5.2' - androidTestImplementation 'commons-net:commons-net:3.10.0' + androidTestImplementation 'commons-net:commons-net:3.9.0' // later version fails on old Android devices with missing java.time.Duration class androidTestImplementation 'org.apache.commons:commons-lang3:3.14.0' } diff --git a/usbSerialForAndroid/src/androidTest/java/com/hoho/android/usbserial/DeviceTest.java b/usbSerialForAndroid/src/androidTest/java/com/hoho/android/usbserial/DeviceTest.java index ca0546dc..df82d5dc 100644 --- a/usbSerialForAndroid/src/androidTest/java/com/hoho/android/usbserial/DeviceTest.java +++ b/usbSerialForAndroid/src/androidTest/java/com/hoho/android/usbserial/DeviceTest.java @@ -1425,19 +1425,16 @@ public void IoManager() throws Exception { usb.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE); telnet.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE); usb.ioManager.setThreadPriority(Process.THREAD_PRIORITY_DEFAULT); + assertEquals(SerialInputOutputManager.State.STOPPED, usb.ioManager.getState()); usb.ioManager.start(); usb.waitForIoManagerStarted(); - assertTrue("iomanager thread", usb.hasIoManagerThread()); + assertEquals(SerialInputOutputManager.State.RUNNING, usb.ioManager.getState()); + assertTrue("iomanager thread", usb.hasIoManagerThreads()); try { usb.ioManager.start(); fail("already running error expected"); } catch (IllegalStateException ignored) { } - try { - usb.ioManager.run(); - fail("already running error expected"); - } catch (IllegalStateException ignored) { - } try { usb.ioManager.setThreadPriority(Process.THREAD_PRIORITY_LOWEST); fail("setThreadPriority IllegalStateException expected"); @@ -1462,23 +1459,12 @@ public void IoManager() throws Exception { telnet.write(new byte[1]); // now uses 8 byte buffer usb.read(3); - // writebuffer resize + // small writebuffer try { usb.ioManager.writeAsync(new byte[8192]); fail("expected BufferOverflowException"); } catch (BufferOverflowException ignored) {} - usb.ioManager.setWriteBufferSize(16); - usb.ioManager.writeAsync("1234567890AB".getBytes()); - try { - usb.ioManager.setWriteBufferSize(8); - fail("expected BufferOverflowException"); - } catch (BufferOverflowException ignored) {} - usb.ioManager.setWriteBufferSize(24); // pending date copied to new buffer - telnet.write("a".getBytes()); - assertThat(usb.read(1), equalTo("a".getBytes())); - assertThat(telnet.read(12), equalTo("1234567890AB".getBytes())); - // small readbuffer usb.ioManager.setReadBufferSize(8); Log.d(TAG, "setReadBufferSize(8)"); @@ -1490,74 +1476,31 @@ public void IoManager() throws Exception { telnet.write("d".getBytes()); assertThat(usb.read(1), equalTo("d".getBytes())); + SerialInputOutputManager ioManager = usb.ioManager; + assertEquals(SerialInputOutputManager.State.RUNNING, usb.ioManager.getState()); usb.close(); - for (int i = 0; i < 100 && usb.hasIoManagerThread(); i++) { + for (int i = 0; i < 100 && usb.hasIoManagerThreads(); i++) { Thread.sleep(1); } - assertFalse("iomanager thread", usb.hasIoManagerThread()); + assertFalse("iomanager threads", usb.hasIoManagerThreads()); + assertNull(usb.ioManager); + assertEquals(SerialInputOutputManager.State.STOPPED, ioManager.getState()); SerialInputOutputManager.DEBUG = false; - - // legacy start - usb.open(EnumSet.of(UsbWrapper.OpenCloseFlags.NO_IOMANAGER_START)); // creates new IoManager - usb.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE); - telnet.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE); - usb.ioManager.setThreadPriority(Process.THREAD_PRIORITY_DEFAULT); - Executors.newSingleThreadExecutor().submit(usb.ioManager); - usb.waitForIoManagerStarted(); - try { - usb.ioManager.start(); - fail("already running error expected"); - } catch (IllegalStateException ignored) { - } } @Test public void writeAsync() throws Exception { byte[] data, buf = new byte[]{1}; - // w/o timeout: write delayed until something is read + // write immediately, without waiting for read usb.open(); usb.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE); telnet.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE); usb.ioManager.writeAsync(buf); usb.ioManager.writeAsync(buf); - data = telnet.read(1); - assertEquals(0, data.length); - telnet.write(buf); - data = usb.read(1); - assertEquals(1, data.length); data = telnet.read(2); assertEquals(2, data.length); usb.close(); - - // with timeout: write after timeout - usb.open(EnumSet.of(UsbWrapper.OpenCloseFlags.NO_IOMANAGER_START)); - usb.ioManager.setReadTimeout(100); - usb.ioManager.start(); - usb.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE); - telnet.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE); - usb.ioManager.writeAsync(buf); - usb.ioManager.writeAsync(buf); - data = telnet.read(2); - assertEquals(2, data.length); - usb.ioManager.setReadTimeout(200); - - // with internal SerialTimeoutException - TestBuffer tbuf = new TestBuffer(usb.writeBufferSize + 2*usb.writePacketSize); - byte[] pbuf1 = new byte[tbuf.buf.length - 4]; - byte[] pbuf2 = new byte[1]; - System.arraycopy(tbuf.buf, 0,pbuf1, 0, pbuf1.length); - usb.ioManager.setWriteTimeout(20); // tbuf len >= 128, needs 133msec @ 9600 baud - usb.setParameters(9600, 8, 1, UsbSerialPort.PARITY_NONE); - telnet.setParameters(9600, 8, 1, UsbSerialPort.PARITY_NONE); - usb.ioManager.writeAsync(pbuf1); - for(int i = pbuf1.length; i < tbuf.buf.length; i++) { - Thread.sleep(20); - pbuf2[0] = tbuf.buf[i]; - usb.ioManager.writeAsync(pbuf2); - } - while(!tbuf.testRead(telnet.read(-1))) - ; } @Test diff --git a/usbSerialForAndroid/src/androidTest/java/com/hoho/android/usbserial/util/UsbWrapper.java b/usbSerialForAndroid/src/androidTest/java/com/hoho/android/usbserial/util/UsbWrapper.java index 3d8e8856..034928dc 100644 --- a/usbSerialForAndroid/src/androidTest/java/com/hoho/android/usbserial/util/UsbWrapper.java +++ b/usbSerialForAndroid/src/androidTest/java/com/hoho/android/usbserial/util/UsbWrapper.java @@ -33,6 +33,8 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import androidx.core.content.ContextCompat; + public class UsbWrapper implements SerialInputOutputManager.Listener { public final static int USB_READ_WAIT = 500; @@ -92,7 +94,7 @@ public void onReceive(Context context, Intent intent) { intent.setPackage(context.getPackageName()); PendingIntent permissionIntent = PendingIntent.getBroadcast(context, 0, intent, flags); IntentFilter filter = new IntentFilter("com.android.example.USB_PERMISSION"); - context.registerReceiver(usbReceiver, filter, Context.RECEIVER_NOT_EXPORTED); + ContextCompat.registerReceiver(context, usbReceiver, filter, Context.RECEIVER_NOT_EXPORTED); usbManager.requestPermission(serialDriver.getDevice(), permissionIntent); for(int i=0; i<5000; i++) { if(granted[0] != null) break; @@ -256,12 +258,15 @@ public void waitForIoManagerStarted() throws IOException { throw new IOException("IoManager not started"); } - public boolean hasIoManagerThread() { + public boolean hasIoManagerThreads() { + int c = 0; for (Thread thread : Thread.getAllStackTraces().keySet()) { - if (thread.getName().equals(SerialInputOutputManager.class.getSimpleName())) - return true; + if (thread.getName().equals(SerialInputOutputManager.class.getSimpleName() + "_read")) + c += 1; + if (thread.getName().equals(SerialInputOutputManager.class.getSimpleName() + "_write")) + c += 1; } - return false; + return c == 2; } // wait full time diff --git a/usbSerialForAndroid/src/main/java/com/hoho/android/usbserial/util/SerialInputOutputManager.java b/usbSerialForAndroid/src/main/java/com/hoho/android/usbserial/util/SerialInputOutputManager.java index 92a12756..c22a2b8c 100644 --- a/usbSerialForAndroid/src/main/java/com/hoho/android/usbserial/util/SerialInputOutputManager.java +++ b/usbSerialForAndroid/src/main/java/com/hoho/android/usbserial/util/SerialInputOutputManager.java @@ -9,21 +9,23 @@ import android.os.Process; import android.util.Log; -import com.hoho.android.usbserial.driver.SerialTimeoutException; import com.hoho.android.usbserial.driver.UsbSerialPort; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicReference; /** - * Utility class which services a {@link UsbSerialPort} in its {@link #run()} method. + * Utility class which services a {@link UsbSerialPort} in its {@link #runWrite()} ()} and {@link #runRead()} ()} ()} methods. * * @author mike wakerly (opensource@hoho.com) */ -public class SerialInputOutputManager implements Runnable { +public class SerialInputOutputManager { public enum State { STOPPED, + STARTING, RUNNING, STOPPING } @@ -33,9 +35,6 @@ public enum State { private static final String TAG = SerialInputOutputManager.class.getSimpleName(); private static final int BUFSIZ = 4096; - /** - * default read timeout is infinite, to avoid data loss with bulkTransfer API - */ private int mReadTimeout = 0; private int mWriteTimeout = 0; @@ -46,7 +45,8 @@ public enum State { private ByteBuffer mWriteBuffer = ByteBuffer.allocate(BUFSIZ); private int mThreadPriority = Process.THREAD_PRIORITY_URGENT_AUDIO; - private State mState = State.STOPPED; // Synchronized by 'this' + private final AtomicReference mState = new AtomicReference<>(State.STOPPED); + private CountDownLatch mStartuplatch = new CountDownLatch(2); private Listener mListener; // Synchronized by 'this' private final UsbSerialPort mSerialPort; @@ -57,7 +57,7 @@ public interface Listener { void onNewData(byte[] data); /** - * Called when {@link SerialInputOutputManager#run()} aborts due to an error. + * Called when {@link SerialInputOutputManager#runRead()} ()} or {@link SerialInputOutputManager#runWrite()} ()} ()} aborts due to an error. */ void onRunError(Exception e); } @@ -87,8 +87,9 @@ public synchronized Listener getListener() { * @param threadPriority see {@link Process#setThreadPriority(int)} * */ public void setThreadPriority(int threadPriority) { - if (mState != State.STOPPED) + if (!mState.compareAndSet(State.STOPPED, State.STOPPED)) { throw new IllegalStateException("threadPriority only configurable before SerialInputOutputManager is started"); + } mThreadPriority = threadPriority; } @@ -97,7 +98,7 @@ public void setThreadPriority(int threadPriority) { */ public void setReadTimeout(int timeout) { // when set if already running, read already blocks and the new value will not become effective now - if(mReadTimeout == 0 && timeout != 0 && mState != State.STOPPED) + if(mReadTimeout == 0 && timeout != 0 && mState.get() != State.STOPPED) throw new IllegalStateException("readTimeout only configurable before SerialInputOutputManager is started"); mReadTimeout = timeout; } @@ -145,91 +146,150 @@ public int getWriteBufferSize() { } /** - * when using writeAsync, it is recommended to use readTimeout != 0, - * else the write will be delayed until read data is available + * write data asynchronously */ public void writeAsync(byte[] data) { synchronized (mWriteBufferLock) { mWriteBuffer.put(data); + mWriteBufferLock.notifyAll(); // Notify waiting threads } } /** - * start SerialInputOutputManager in separate thread + * start SerialInputOutputManager in separate threads */ public void start() { - if(mState != State.STOPPED) + if(mState.compareAndSet(State.STOPPED, State.STARTING)) { + mStartuplatch = new CountDownLatch(2); + new Thread(this::runRead, this.getClass().getSimpleName() + "_read").start(); + new Thread(this::runWrite, this.getClass().getSimpleName() + "_write").start(); + try { + mStartuplatch.await(); + mState.set(State.RUNNING); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } else { throw new IllegalStateException("already started"); - new Thread(this, this.getClass().getSimpleName()).start(); + } } /** - * stop SerialInputOutputManager thread + * stop SerialInputOutputManager threads * * when using readTimeout == 0 (default), additionally use usbSerialPort.close() to * interrupt blocking read */ - public synchronized void stop() { - if (getState() == State.RUNNING) { + public void stop() { + if(mState.compareAndSet(State.RUNNING, State.STOPPING)) { + synchronized (mWriteBufferLock) { + mWriteBufferLock.notifyAll(); // Wake up any waiting thread to check the stop condition + } Log.i(TAG, "Stop requested"); - mState = State.STOPPING; } } - public synchronized State getState() { - return mState; + public State getState() { + return mState.get(); } /** - * Continuously services the read and write buffers until {@link #stop()} is - * called, or until a driver exception is raised. + * @return true if the thread is still running */ - @Override - public void run() { - synchronized (this) { - if (getState() != State.STOPPED) { - throw new IllegalStateException("Already running"); + private boolean isStillRunning() { + State state = mState.get(); + return ((state == State.RUNNING) || (state == State.STARTING)) + && !Thread.currentThread().isInterrupted(); + } + + /** + * Notify listener of an error + * + * @param e the exception + */ + private void notifyErrorListener(Throwable e) { + Listener listener = getListener(); + if (listener != null) { + try { + listener.onRunError(e instanceof Exception ? (Exception) e : new Exception(e)); + } catch (Throwable t) { + Log.w(TAG, "Exception in onRunError: " + t.getMessage(), t); } - mState = State.RUNNING; } - Log.i(TAG, "Running ..."); + } + + /** + * Set the thread priority + */ + private void setThreadPriority() { + if (mThreadPriority != Process.THREAD_PRIORITY_DEFAULT) { + Process.setThreadPriority(mThreadPriority); + } + } + + /** + * Continuously services the read buffers until {@link #stop()} is called, or until a driver exception is + * raised. + */ + void runRead() { + Log.i(TAG, "runRead running ..."); try { - if(mThreadPriority != Process.THREAD_PRIORITY_DEFAULT) - Process.setThreadPriority(mThreadPriority); - while (true) { - if (getState() != State.RUNNING) { - Log.i(TAG, "Stopping mState=" + getState()); - break; - } - step(); - } + setThreadPriority(); + mStartuplatch.countDown(); + do { + stepRead(); + } while (isStillRunning()); + Log.i(TAG, "runRead: Stopping mState=" + getState()); } catch (Throwable e) { - if(mSerialPort.isOpen()) { - Log.w(TAG, "Run ending due to exception: " + e.getMessage(), e); + if (Thread.currentThread().isInterrupted()) { + Log.w(TAG, "runRead: interrupted"); + } else if(mSerialPort.isOpen()) { + Log.w(TAG, "runRead ending due to exception: " + e.getMessage(), e); } else { - Log.i(TAG, "Socket closed"); + Log.i(TAG, "runRead: Socket closed"); } - final Listener listener = getListener(); - if (listener != null) { - try { - if (e instanceof Exception) { - listener.onRunError((Exception) e); - } else { - listener.onRunError(new Exception(e)); - } - } catch (Throwable t) { - Log.w(TAG, "Exception in onRunError: " + t.getMessage(), t); + notifyErrorListener(e); + } finally { + if (!mState.compareAndSet(State.RUNNING, State.STOPPING)) { + if (mState.compareAndSet(State.STOPPING, State.STOPPED)) { + Log.i(TAG, "runRead: Stopped mState=" + getState()); } } + } + } + + /** + * Continuously services the write buffers until {@link #stop()} is called, or until a driver exception is + * raised. + */ + void runWrite() { + Log.i(TAG, "runWrite running ..."); + try { + setThreadPriority(); + mStartuplatch.countDown(); + do { + stepWrite(); + } while (isStillRunning()); + Log.i(TAG, "runWrite: Stopping mState=" + getState()); + } catch (Throwable e) { + if (Thread.currentThread().isInterrupted()) { + Log.w(TAG, "runWrite: interrupted"); + } else if(mSerialPort.isOpen()) { + Log.w(TAG, "runWrite ending due to exception: " + e.getMessage(), e); + } else { + Log.i(TAG, "runWrite: Socket closed"); + } + notifyErrorListener(e); } finally { - synchronized (this) { - mState = State.STOPPED; - Log.i(TAG, "Stopped"); + if (!mState.compareAndSet(State.RUNNING, State.STOPPING)) { + if (mState.compareAndSet(State.STOPPING, State.STOPPED)) { + Log.i(TAG, "runWrite: Stopped mState=" + getState()); + } } } } - private void step() throws IOException { + private void stepRead() throws IOException { // Handle incoming data. byte[] buffer; synchronized (mReadBufferLock) { @@ -247,39 +307,28 @@ private void step() throws IOException { listener.onNewData(data); } } + } + private void stepWrite() throws IOException, InterruptedException { // Handle outgoing data. - buffer = null; + byte[] buffer = null; synchronized (mWriteBufferLock) { - len = mWriteBuffer.position(); + int len = mWriteBuffer.position(); if (len > 0) { buffer = new byte[len]; mWriteBuffer.rewind(); mWriteBuffer.get(buffer, 0, len); mWriteBuffer.clear(); + mWriteBufferLock.notifyAll(); // Notify writeAsync that there is space in the buffer + } else { + mWriteBufferLock.wait(); } } if (buffer != null) { if (DEBUG) { - Log.d(TAG, "Writing data len=" + len); - } - try { - mSerialPort.write(buffer, mWriteTimeout); - } catch (SerialTimeoutException ex) { - synchronized (mWriteBufferLock) { - byte[] buffer2 = null; - int len2 = mWriteBuffer.position(); - if (len2 > 0) { - buffer2 = new byte[len2]; - mWriteBuffer.rewind(); - mWriteBuffer.get(buffer2, 0, len2); - mWriteBuffer.clear(); - } - mWriteBuffer.put(buffer, ex.bytesTransferred, buffer.length - ex.bytesTransferred); - if (buffer2 != null) - mWriteBuffer.put(buffer2); - } + Log.d(TAG, "Writing data len=" + buffer.length); } + mSerialPort.write(buffer, mWriteTimeout); } } diff --git a/usbSerialForAndroid/src/test/java/com/hoho/android/usbserial/util/SerialInputOutputManagerTest.java b/usbSerialForAndroid/src/test/java/com/hoho/android/usbserial/util/SerialInputOutputManagerTest.java index 206aadc3..63d9f209 100644 --- a/usbSerialForAndroid/src/test/java/com/hoho/android/usbserial/util/SerialInputOutputManagerTest.java +++ b/usbSerialForAndroid/src/test/java/com/hoho/android/usbserial/util/SerialInputOutputManagerTest.java @@ -34,18 +34,19 @@ class ErrorListener implements SerialInputOutputManager.Listener { CommonUsbSerialPort port = mock(CommonUsbSerialPort.class); when(port.getReadEndpoint()).thenReturn(readEndpoint); when(port.read(new byte[16], 0)).thenReturn(1); + when(port.isOpen()).thenReturn(true); SerialInputOutputManager manager = new SerialInputOutputManager(port); manager.setThreadPriority(Process.THREAD_PRIORITY_DEFAULT); ExceptionListener exceptionListener = new ExceptionListener(); manager.setListener(exceptionListener); - manager.run(); + manager.runRead(); assertEquals(RuntimeException.class, exceptionListener.e.getClass()); assertEquals("exception1", exceptionListener.e.getMessage()); ErrorListener errorListener = new ErrorListener(); manager.setListener(errorListener); - manager.run(); + manager.runRead(); assertEquals(Exception.class, errorListener.e.getClass()); assertEquals("java.lang.UnknownError: error1", errorListener.e.getMessage()); assertEquals(UnknownError.class, errorListener.e.getCause().getClass());