tests/AccessoryDisplay/common/src/com/android/accessorydisplay/common/Transport.java - platform/frameworks/base - Git at Google

 /*
  * Copyright (C) 2013 The Android Open Source Project
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package com.android.accessorydisplay.common;

 import android.os.Handler;
 import android.os.Looper;
 import android.os.Message;
 import android.util.SparseArray;

 import java.io.IOException;
 import java.nio.ByteBuffer;

 /**
  * A simple message transport.
  * <p>
  * This object's interface is thread-safe, however incoming messages
  * are always delivered on the {@link Looper} thread on which the transport
  * was created.
  * </p>
  */
 public abstract class Transport {
     private static final int MAX_INPUT_BUFFERS = 8;

     private final Logger mLogger;

     // The transport thread looper and handler.
     private final TransportHandler mHandler;

     // Lock to guard all mutable state.
     private final Object mLock = new Object();

     // The output buffer.  Set to null when the transport is closed.
     private ByteBuffer mOutputBuffer;

     // The input buffer pool.
     private BufferPool mInputBufferPool;

     // The reader thread.  Initialized when reading starts.
     private ReaderThread mThread;

     // The list of callbacks indexed by service id.
     private final SparseArray<Callback> mServices = new SparseArray<Callback>();

     public Transport(Logger logger, int maxPacketSize) {
         mLogger = logger;
         mHandler = new TransportHandler();
         mOutputBuffer = ByteBuffer.allocate(maxPacketSize);
         mInputBufferPool = new BufferPool(
                 maxPacketSize, Protocol.MAX_ENVELOPE_SIZE, MAX_INPUT_BUFFERS);
     }

     /**
      * Gets the logger for debugging.
      */
     public Logger getLogger() {
         return mLogger;
     }

     /**
      * Gets the handler on the transport's thread.
      */
     public Handler getHandler() {
         return mHandler;
     }

     /**
      * Closes the transport.
      */
     public void close() {
         synchronized (mLock) {
             if (mOutputBuffer != null) {
                 if (mThread == null) {
                     ioClose();
                 } else {
                     // If the thread was started then it will be responsible for
                     // closing the stream when it quits because it may currently
                     // be in the process of reading from the stream so we can't simply
                     // shut it down right now.
                     mThread.quit();
                 }
                 mOutputBuffer = null;
             }
         }
     }

     /**
      * Sends a message.
      *
      * @param service The service to whom the message is addressed.
      * @param what The message type.
      * @param content The content, or null if there is none.
      * @return True if the message was sent successfully, false if an error occurred.
      */
     public boolean sendMessage(int service, int what, ByteBuffer content) {
         checkServiceId(service);
         checkMessageId(what);

         try {
             synchronized (mLock) {
                 if (mOutputBuffer == null) {
                     mLogger.logError("Send message failed because transport was closed.");
                     return false;
                 }

                 final byte[] outputArray = mOutputBuffer.array();
                 final int capacity = mOutputBuffer.capacity();
                 mOutputBuffer.clear();
                 mOutputBuffer.putShort((short)service);
                 mOutputBuffer.putShort((short)what);
                 if (content == null) {
                     mOutputBuffer.putInt(0);
                 } else {
                     final int contentLimit = content.limit();
                     int contentPosition = content.position();
                     int contentRemaining = contentLimit - contentPosition;
                     if (contentRemaining > Protocol.MAX_CONTENT_SIZE) {
                         throw new IllegalArgumentException("Message content too large: "
                                 + contentRemaining + " > " + Protocol.MAX_CONTENT_SIZE);
                     }
                     mOutputBuffer.putInt(contentRemaining);
                     while (contentRemaining != 0) {
                         final int outputAvailable = capacity - mOutputBuffer.position();
                         if (contentRemaining <= outputAvailable) {
                             mOutputBuffer.put(content);
                             break;
                         }
                         content.limit(contentPosition + outputAvailable);
                         mOutputBuffer.put(content);
                         content.limit(contentLimit);
                         ioWrite(outputArray, 0, capacity);
                         contentPosition += outputAvailable;
                         contentRemaining -= outputAvailable;
                         mOutputBuffer.clear();
                     }
                 }
                 ioWrite(outputArray, 0, mOutputBuffer.position());
                 return true;
             }
         } catch (IOException ex) {
             mLogger.logError("Send message failed: " + ex);
             return false;
         }
     }

     /**
      * Starts reading messages on a separate thread.
      */
     public void startReading() {
         synchronized (mLock) {
             if (mOutputBuffer == null) {
                 throw new IllegalStateException("Transport has been closed");
             }

             mThread = new ReaderThread();
             mThread.start();
         }
     }

     /**
      * Registers a service and provides a callback to receive messages.
      *
      * @param service The service id.
      * @param callback The callback to use.
      */
     public void registerService(int service, Callback callback) {
         checkServiceId(service);
         if (callback == null) {
             throw new IllegalArgumentException("callback must not be null");
         }

         synchronized (mLock) {
             mServices.put(service, callback);
         }
     }

     /**
      * Unregisters a service.
      *
      * @param service The service to unregister.
      */
     public void unregisterService(int service) {
         checkServiceId(service);

         synchronized (mLock) {
             mServices.remove(service);
         }
     }

     private void dispatchMessageReceived(int service, int what, ByteBuffer content) {
         final Callback callback;
         synchronized (mLock) {
             callback = mServices.get(service);
         }
         if (callback != null) {
             callback.onMessageReceived(service, what, content);
         } else {
             mLogger.log("Discarding message " + what
                     + " for unregistered service " + service);
         }
     }

     private static void checkServiceId(int service) {
         if (service < 0 || service > 0xffff) {
             throw new IllegalArgumentException("service id out of range: " + service);
         }
     }

     private static void checkMessageId(int what) {
         if (what < 0 || what > 0xffff) {
             throw new IllegalArgumentException("message id out of range: " + what);
         }
     }

     // The IO methods must be safe to call on any thread.
     // They may be called concurrently.
     protected abstract void ioClose();
     protected abstract int ioRead(byte[] buffer, int offset, int count)
             throws IOException;
     protected abstract void ioWrite(byte[] buffer, int offset, int count)
             throws IOException;

     /**
      * Callback for services that handle received messages.
      */
     public interface Callback {
         /**
          * Indicates that a message was received.
          *
          * @param service The service to whom the message is addressed.
          * @param what The message type.
          * @param content The content, or null if there is none.
          */
         public void onMessageReceived(int service, int what, ByteBuffer content);
     }

     final class TransportHandler extends Handler {
         @Override
         public void handleMessage(Message msg) {
             final ByteBuffer buffer = (ByteBuffer)msg.obj;
             try {
                 final int limit = buffer.limit();
                 while (buffer.position() < limit) {
                     final int service = buffer.getShort() & 0xffff;
                     final int what = buffer.getShort() & 0xffff;
                     final int contentSize = buffer.getInt();
                     if (contentSize == 0) {
                         dispatchMessageReceived(service, what, null);
                     } else {
                         final int end = buffer.position() + contentSize;
                         buffer.limit(end);
                         dispatchMessageReceived(service, what, buffer);
                         buffer.limit(limit);
                         buffer.position(end);
                     }
                 }
             } finally {
                 mInputBufferPool.release(buffer);
             }
         }
     }

     final class ReaderThread extends Thread {
         // Set to true when quitting.
         private volatile boolean mQuitting;

         public ReaderThread() {
             super("Accessory Display Transport");
         }

         @Override
         public void run() {
             loop();
             ioClose();
         }

         private void loop() {
             ByteBuffer buffer = null;
             int length = Protocol.HEADER_SIZE;
             int contentSize = -1;
             outer: while (!mQuitting) {
                 // Get a buffer.
                 if (buffer == null) {
                     buffer = mInputBufferPool.acquire(length);
                 } else {
                     buffer = mInputBufferPool.grow(buffer, length);
                 }

                 // Read more data until needed number of bytes obtained.
                 int position = buffer.position();
                 int count;
                 try {
                     count = ioRead(buffer.array(), position, buffer.capacity() - position);
                     if (count < 0) {
                         break; // end of stream
                     }
                 } catch (IOException ex) {
                     mLogger.logError("Read failed: " + ex);
                     break; // error
                 }
                 position += count;
                 buffer.position(position);
                 if (contentSize < 0 && position >= Protocol.HEADER_SIZE) {
                     contentSize = buffer.getInt(4);
                     if (contentSize < 0 || contentSize > Protocol.MAX_CONTENT_SIZE) {
                         mLogger.logError("Encountered invalid content size: " + contentSize);
                         break; // malformed stream
                     }
                     length += contentSize;
                 }
                 if (position < length) {
                     continue; // need more data
                 }

                 // There is at least one complete message in the buffer.
                 // Find the end of a contiguous chunk of complete messages.
                 int next = length;
                 int remaining;
                 for (;;) {
                     length = Protocol.HEADER_SIZE;
                     remaining = position - next;
                     if (remaining < length) {
                         contentSize = -1;
                         break; // incomplete header, need more data
                     }
                     contentSize = buffer.getInt(next + 4);
                     if (contentSize < 0 || contentSize > Protocol.MAX_CONTENT_SIZE) {
                         mLogger.logError("Encountered invalid content size: " + contentSize);
                         break outer; // malformed stream
                     }
                     length += contentSize;
                     if (remaining < length) {
                         break; // incomplete content, need more data
                     }
                     next += length;
                 }

                 // Post the buffer then don't modify it anymore.
                 // Now this is kind of sneaky.  We know that no other threads will
                 // be acquiring buffers from the buffer pool so we can keep on
                 // referring to this buffer as long as we don't modify its contents.
                 // This allows us to operate in a single-buffered mode if desired.
                 buffer.limit(next);
                 buffer.rewind();
                 mHandler.obtainMessage(0, buffer).sendToTarget();

                 // If there is an incomplete message at the end, then we will need
                 // to copy it to a fresh buffer before continuing.  In the single-buffered
                 // case, we may acquire the same buffer as before which is fine.
                 if (remaining == 0) {
                     buffer = null;
                 } else {
                     final ByteBuffer oldBuffer = buffer;
                     buffer = mInputBufferPool.acquire(length);
                     System.arraycopy(oldBuffer.array(), next, buffer.array(), 0, remaining);
                     buffer.position(remaining);
                 }
             }

             if (buffer != null) {
                 mInputBufferPool.release(buffer);
             }
         }

         public void quit() {
             mQuitting = true;
         }
     }
 }
	/*
	* Copyright (C) 2013 The Android Open Source Project
	*
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package com.android.accessorydisplay.common;

	import android.os.Handler;
	import android.os.Looper;
	import android.os.Message;
	import android.util.SparseArray;

	import java.io.IOException;
	import java.nio.ByteBuffer;

	/**
	* A simple message transport.
	* <p>
	* This object's interface is thread-safe, however incoming messages
	* are always delivered on the {@link Looper} thread on which the transport
	* was created.
	* </p>
	*/
	public abstract class Transport {
	private static final int MAX_INPUT_BUFFERS = 8;

	private final Logger mLogger;

	// The transport thread looper and handler.
	private final TransportHandler mHandler;

	// Lock to guard all mutable state.
	private final Object mLock = new Object();

	// The output buffer. Set to null when the transport is closed.
	private ByteBuffer mOutputBuffer;

	// The input buffer pool.
	private BufferPool mInputBufferPool;

	// The reader thread. Initialized when reading starts.
	private ReaderThread mThread;

	// The list of callbacks indexed by service id.
	private final SparseArray<Callback> mServices = new SparseArray<Callback>();

	public Transport(Logger logger, int maxPacketSize) {
	mLogger = logger;
	mHandler = new TransportHandler();
	mOutputBuffer = ByteBuffer.allocate(maxPacketSize);
	mInputBufferPool = new BufferPool(
	maxPacketSize, Protocol.MAX_ENVELOPE_SIZE, MAX_INPUT_BUFFERS);
	}

	/**
	* Gets the logger for debugging.
	*/
	public Logger getLogger() {
	return mLogger;
	}

	/**
	* Gets the handler on the transport's thread.
	*/
	public Handler getHandler() {
	return mHandler;
	}

	/**
	* Closes the transport.
	*/
	public void close() {
	synchronized (mLock) {
	if (mOutputBuffer != null) {
	if (mThread == null) {
	ioClose();
	} else {
	// If the thread was started then it will be responsible for
	// closing the stream when it quits because it may currently
	// be in the process of reading from the stream so we can't simply
	// shut it down right now.
	mThread.quit();
	}
	mOutputBuffer = null;
	}
	}
	}

	/**
	* Sends a message.
	*
	* @param service The service to whom the message is addressed.
	* @param what The message type.
	* @param content The content, or null if there is none.
	* @return True if the message was sent successfully, false if an error occurred.
	*/
	public boolean sendMessage(int service, int what, ByteBuffer content) {
	checkServiceId(service);
	checkMessageId(what);

	try {
	synchronized (mLock) {
	if (mOutputBuffer == null) {
	mLogger.logError("Send message failed because transport was closed.");
	return false;
	}

	final byte[] outputArray = mOutputBuffer.array();
	final int capacity = mOutputBuffer.capacity();
	mOutputBuffer.clear();
	mOutputBuffer.putShort((short)service);
	mOutputBuffer.putShort((short)what);
	if (content == null) {
	mOutputBuffer.putInt(0);
	} else {
	final int contentLimit = content.limit();
	int contentPosition = content.position();
	int contentRemaining = contentLimit - contentPosition;
	if (contentRemaining > Protocol.MAX_CONTENT_SIZE) {
	throw new IllegalArgumentException("Message content too large: "
	+ contentRemaining + " > " + Protocol.MAX_CONTENT_SIZE);
	}
	mOutputBuffer.putInt(contentRemaining);
	while (contentRemaining != 0) {
	final int outputAvailable = capacity - mOutputBuffer.position();
	if (contentRemaining <= outputAvailable) {
	mOutputBuffer.put(content);
	break;
	}
	content.limit(contentPosition + outputAvailable);
	mOutputBuffer.put(content);
	content.limit(contentLimit);
	ioWrite(outputArray, 0, capacity);
	contentPosition += outputAvailable;
	contentRemaining -= outputAvailable;
	mOutputBuffer.clear();
	}
	}
	ioWrite(outputArray, 0, mOutputBuffer.position());
	return true;
	}
	} catch (IOException ex) {
	mLogger.logError("Send message failed: " + ex);
	return false;
	}
	}

	/**
	* Starts reading messages on a separate thread.
	*/
	public void startReading() {
	synchronized (mLock) {
	if (mOutputBuffer == null) {
	throw new IllegalStateException("Transport has been closed");
	}

	mThread = new ReaderThread();
	mThread.start();
	}
	}

	/**
	* Registers a service and provides a callback to receive messages.
	*
	* @param service The service id.
	* @param callback The callback to use.
	*/
	public void registerService(int service, Callback callback) {
	checkServiceId(service);
	if (callback == null) {
	throw new IllegalArgumentException("callback must not be null");
	}

	synchronized (mLock) {
	mServices.put(service, callback);
	}
	}

	/**
	* Unregisters a service.
	*
	* @param service The service to unregister.
	*/
	public void unregisterService(int service) {
	checkServiceId(service);

	synchronized (mLock) {
	mServices.remove(service);
	}
	}

	private void dispatchMessageReceived(int service, int what, ByteBuffer content) {
	final Callback callback;
	synchronized (mLock) {
	callback = mServices.get(service);
	}
	if (callback != null) {
	callback.onMessageReceived(service, what, content);
	} else {
	mLogger.log("Discarding message " + what
	+ " for unregistered service " + service);
	}
	}

	private static void checkServiceId(int service) {
	if (service < 0 \|\| service > 0xffff) {
	throw new IllegalArgumentException("service id out of range: " + service);
	}
	}

	private static void checkMessageId(int what) {
	if (what < 0 \|\| what > 0xffff) {
	throw new IllegalArgumentException("message id out of range: " + what);
	}
	}

	// The IO methods must be safe to call on any thread.
	// They may be called concurrently.
	protected abstract void ioClose();
	protected abstract int ioRead(byte[] buffer, int offset, int count)
	throws IOException;
	protected abstract void ioWrite(byte[] buffer, int offset, int count)
	throws IOException;

	/**
	* Callback for services that handle received messages.
	*/
	public interface Callback {
	/**
	* Indicates that a message was received.
	*
	* @param service The service to whom the message is addressed.
	* @param what The message type.
	* @param content The content, or null if there is none.
	*/
	public void onMessageReceived(int service, int what, ByteBuffer content);
	}

	final class TransportHandler extends Handler {
	@Override
	public void handleMessage(Message msg) {
	final ByteBuffer buffer = (ByteBuffer)msg.obj;
	try {
	final int limit = buffer.limit();
	while (buffer.position() < limit) {
	final int service = buffer.getShort() & 0xffff;
	final int what = buffer.getShort() & 0xffff;
	final int contentSize = buffer.getInt();
	if (contentSize == 0) {
	dispatchMessageReceived(service, what, null);
	} else {
	final int end = buffer.position() + contentSize;
	buffer.limit(end);
	dispatchMessageReceived(service, what, buffer);
	buffer.limit(limit);
	buffer.position(end);
	}
	}
	} finally {
	mInputBufferPool.release(buffer);
	}
	}
	}

	final class ReaderThread extends Thread {
	// Set to true when quitting.
	private volatile boolean mQuitting;

	public ReaderThread() {
	super("Accessory Display Transport");
	}

	@Override
	public void run() {
	loop();
	ioClose();
	}

	private void loop() {
	ByteBuffer buffer = null;
	int length = Protocol.HEADER_SIZE;
	int contentSize = -1;
	outer: while (!mQuitting) {
	// Get a buffer.
	if (buffer == null) {
	buffer = mInputBufferPool.acquire(length);
	} else {
	buffer = mInputBufferPool.grow(buffer, length);
	}

	// Read more data until needed number of bytes obtained.
	int position = buffer.position();
	int count;
	try {
	count = ioRead(buffer.array(), position, buffer.capacity() - position);
	if (count < 0) {
	break; // end of stream
	}
	} catch (IOException ex) {
	mLogger.logError("Read failed: " + ex);
	break; // error
	}
	position += count;
	buffer.position(position);
	if (contentSize < 0 && position >= Protocol.HEADER_SIZE) {
	contentSize = buffer.getInt(4);
	if (contentSize < 0 \|\| contentSize > Protocol.MAX_CONTENT_SIZE) {
	mLogger.logError("Encountered invalid content size: " + contentSize);
	break; // malformed stream
	}
	length += contentSize;
	}
	if (position < length) {
	continue; // need more data
	}

	// There is at least one complete message in the buffer.
	// Find the end of a contiguous chunk of complete messages.
	int next = length;
	int remaining;
	for (;;) {
	length = Protocol.HEADER_SIZE;
	remaining = position - next;
	if (remaining < length) {
	contentSize = -1;
	break; // incomplete header, need more data
	}
	contentSize = buffer.getInt(next + 4);
	if (contentSize < 0 \|\| contentSize > Protocol.MAX_CONTENT_SIZE) {
	mLogger.logError("Encountered invalid content size: " + contentSize);
	break outer; // malformed stream
	}
	length += contentSize;
	if (remaining < length) {
	break; // incomplete content, need more data
	}
	next += length;
	}

	// Post the buffer then don't modify it anymore.
	// Now this is kind of sneaky. We know that no other threads will
	// be acquiring buffers from the buffer pool so we can keep on
	// referring to this buffer as long as we don't modify its contents.
	// This allows us to operate in a single-buffered mode if desired.
	buffer.limit(next);
	buffer.rewind();
	mHandler.obtainMessage(0, buffer).sendToTarget();

	// If there is an incomplete message at the end, then we will need
	// to copy it to a fresh buffer before continuing. In the single-buffered
	// case, we may acquire the same buffer as before which is fine.
	if (remaining == 0) {
	buffer = null;
	} else {
	final ByteBuffer oldBuffer = buffer;
	buffer = mInputBufferPool.acquire(length);
	System.arraycopy(oldBuffer.array(), next, buffer.array(), 0, remaining);
	buffer.position(remaining);
	}
	}

	if (buffer != null) {
	mInputBufferPool.release(buffer);
	}
	}

	public void quit() {
	mQuitting = true;
	}
	}
	}