src/com/android/im/imps/SmsDataChannel.java - platform/packages/apps/IM - Git at Google

 /*
  * Copyright (C) 2008 Esmertec AG.
  * Copyright (C) 2008 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.im.imps;

 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.concurrent.LinkedBlockingQueue;

 import android.os.SystemClock;

 import com.android.im.engine.HeartbeatService;
 import com.android.im.engine.ImErrorInfo;
 import com.android.im.engine.ImException;
 import com.android.im.engine.SmsService;
 import com.android.im.engine.SystemService;
 import com.android.im.engine.SmsService.SmsListener;
 import com.android.im.engine.SmsService.SmsSendFailureCallback;

 public class SmsDataChannel extends DataChannel
         implements SmsListener, HeartbeatService.Callback {
     private SmsService mSmsService;
     private String mSmsAddr;
     private short mSmsPort;

     private long mLastActive;

     private SmsSplitter mSplitter;
     private SmsAssembler mAssembler;

     private LinkedBlockingQueue<Primitive> mReceiveQueue;

     private ImpsTransactionManager mTxManager;
     private boolean mConnected;
     private long mKeepAliveMillis;
     private Primitive mKeepAlivePrimitive;

     private long mReplyTimeout;
     private LinkedList<PendingTransaction> mPendingTransactions;
     private Timer mTimer;

     protected SmsDataChannel(ImpsConnection connection) throws ImException {
         super(connection);

         mTxManager = connection.getTransactionManager();

         ImpsConnectionConfig config = connection.getConfig();
         mReplyTimeout = config.getReplyTimeout();
         mSmsAddr = config.getSmsAddr();
         mSmsPort = (short) config.getSmsPort();
         mSmsService = SystemService.getDefault().getSmsService();

         mParser = new PtsPrimitiveParser();
         try {
             mSerializer = new PtsPrimitiveSerializer(config.getImpsVersion());
         } catch (SerializerException e) {
             throw new ImException(e);
         }
         mSplitter = new SmsSplitter(mSmsService.getMaxSmsLength());
         mAssembler = new SmsAssembler();
         mAssembler.setSmsListener(this);
     }

     @Override
     public void connect() throws ImException {
         mSmsService.addSmsListener(mSmsAddr, mSmsPort, mAssembler);
         mReceiveQueue = new LinkedBlockingQueue<Primitive>();
         mPendingTransactions = new LinkedList<PendingTransaction>();
         mTimer = new Timer(mReplyTimeout);
         new Thread(mTimer, "SmsDataChannel timer").start();
         mConnected = true;
     }

     @Override
     public long getLastActiveTime() {
         return mLastActive;
     }

     @Override
     public boolean isSendingQueueEmpty() {
         // Always true since we don't have a sending queue.
         return true;
     }

     @Override
     public Primitive receivePrimitive() throws InterruptedException {
         return mReceiveQueue.take();
     }

     @Override
     public void sendPrimitive(Primitive p) {
         ByteArrayOutputStream out = new ByteArrayOutputStream();
         try {
             mSerializer.serialize(p, out);
             mSplitter.split(out.toByteArray());
             SmsService smsService =  SystemService.getDefault().getSmsService();
             SendFailureCallback sendFailureCallback
                     = new SendFailureCallback(p.getTransactionID());
             while (mSplitter.hasNext()) {
                 smsService.sendSms(mSmsAddr, mSmsPort, mSplitter.getNext(),
                         sendFailureCallback);
             }
             mLastActive = SystemClock.elapsedRealtime();
             addPendingTransaction(p.getTransactionID());
         } catch (IOException e) {
             mTxManager.notifyErrorResponse(p.getTransactionID(),
                     ImpsErrorInfo.SERIALIZER_ERROR, e.getLocalizedMessage());
         } catch (SerializerException e) {
             mTxManager.notifyErrorResponse(p.getTransactionID(),
                     ImpsErrorInfo.SERIALIZER_ERROR, e.getLocalizedMessage());
         }
     }

     @Override
     public void shutdown() {
         mSmsService.removeSmsListener(this);
         mTimer.stop();
         mConnected = false;
     }

     @Override
     public void startKeepAlive(long interval) {
         if (!mConnected) {
             throw new IllegalStateException();
         }

         if (interval <= 0) {
             interval = mConnection.getConfig().getDefaultKeepAliveInterval();
         }

         mKeepAliveMillis = interval * 1000;
         if (mKeepAliveMillis < 0) {
             ImpsLog.log("Negative keep alive time. Won't send keep-alive");
         }
         mKeepAlivePrimitive = new Primitive(ImpsTags.KeepAlive_Request);

         HeartbeatService heartbeatService
             = SystemService.getDefault().getHeartbeatService();
         if (heartbeatService != null) {
             heartbeatService.startHeartbeat(this, mKeepAliveMillis);
         }
     }

     public long sendHeartbeat() {
         if (!mConnected) {
             return 0;
         }

         long inactiveTime = SystemClock.elapsedRealtime() - mLastActive;
         if (needSendKeepAlive(inactiveTime)) {
             sendKeepAlive();
             return mKeepAliveMillis;
         } else {
             return mKeepAliveMillis - inactiveTime;
         }
     }

     private void sendKeepAlive() {
         ImpsTransactionManager tm = mConnection.getTransactionManager();
         AsyncTransaction tx = new AsyncTransaction(tm) {
             @Override
             public void onResponseError(ImpsErrorInfo error) {
             }

             @Override
             public void onResponseOk(Primitive response) {
                 // Since we never request a new timeout value, the response
                 // can be ignored
             }
         };
         tx.sendRequest(mKeepAlivePrimitive);
     }

     private boolean needSendKeepAlive(long inactiveTime) {
         return mKeepAliveMillis - inactiveTime <= 500;
     }

     @Override
     public boolean resume() {
         return true;
     }

     @Override
     public void suspend() {
         // do nothing.
     }

     public void onIncomingSms(byte[] data) {
         try {
             Primitive p = mParser.parse(new ByteArrayInputStream(data));
             mReceiveQueue.put(p);
             removePendingTransaction(p.getTransactionID());
         } catch (ParserException e) {
             handleError(data, ImpsErrorInfo.PARSER_ERROR, e.getLocalizedMessage());
         } catch (IOException e) {
             handleError(data, ImpsErrorInfo.PARSER_ERROR, e.getLocalizedMessage());
         } catch (InterruptedException e) {
             handleError(data, ImpsErrorInfo.UNKNOWN_ERROR, e.getLocalizedMessage());
         }
     }

     private void handleError(byte[] data, int errCode, String info) {
         String trId = extractTrId(data);
         if (trId != null) {
             mTxManager.notifyErrorResponse(trId, errCode, info);
             removePendingTransaction(trId);
         }
     }

     private String extractTrId(byte[] data) {
         int transIdStart = 4;
         int index = transIdStart;
         while(Character.isDigit(data[index])) {
             index++;
         }
         int transIdLen = index - transIdStart;
         try {
             return new String(data, transIdStart, transIdLen, "UTF-8");
         } catch (UnsupportedEncodingException e) {
             return null;
         }
     }

     private void addPendingTransaction(String transId) {
         synchronized (mPendingTransactions) {
             mPendingTransactions.add(new PendingTransaction(transId));
         }
     }

     private void removePendingTransaction(String transId) {
         synchronized (mPendingTransactions) {
             Iterator<PendingTransaction> iter = mPendingTransactions.iterator();
             while (iter.hasNext()) {
                 PendingTransaction tx = iter.next();
                 if (tx.mTransId.equals(transId)) {
                     iter.remove();
                     break;
                 }
             }
         }
     }

     /*package*/void checkTimeout() {
         synchronized (mPendingTransactions) {
             Iterator<PendingTransaction> iter = mPendingTransactions.iterator();
             while (iter.hasNext()) {
                 PendingTransaction tx = iter.next();
                 if (tx.isExpired(mReplyTimeout)) {
                     notifyTimeout(tx);
                 } else {
                     break;
                 }
             }
         }
     }

     private void notifyTimeout(PendingTransaction tx) {
         String transId = tx.mTransId;
         mTxManager.notifyErrorResponse(transId, ImpsErrorInfo.TIMEOUT,
                 "Timeout");
         removePendingTransaction(transId);
     }

     private class SendFailureCallback implements SmsSendFailureCallback {
         private String mTransId;

         public SendFailureCallback(String transId) {
             mTransId = transId;
         }

         public void onFailure(int errorCode) {
             mTxManager.notifyErrorResponse(mTransId, ImErrorInfo.NETWORK_ERROR, null);
         }
     }

     private class Timer implements Runnable {
         private boolean mStopped;
         private long mInterval;

         public Timer(long interval) {
             mInterval = interval;
             mStopped = false;
         }

         public void stop() {
             mStopped = true;
         }

         public void run() {
             while (!mStopped) {
                 try {
                     Thread.sleep(mInterval);
                 } catch (InterruptedException e) {
                     continue;
                 }
                 checkTimeout();
             }
         }
     }

     private static class PendingTransaction {
         private String mTransId;
         private long mSentTime;

         public PendingTransaction(String transId) {
             mTransId = transId;
         }

         public boolean isExpired(long timeout) {
             return SystemClock.elapsedRealtime() - mSentTime >= timeout;
         }
     }
 }
	/*
	* Copyright (C) 2008 Esmertec AG.
	* Copyright (C) 2008 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.im.imps;

	import java.io.ByteArrayInputStream;
	import java.io.ByteArrayOutputStream;
	import java.io.IOException;
	import java.io.UnsupportedEncodingException;
	import java.util.Iterator;
	import java.util.LinkedList;
	import java.util.concurrent.LinkedBlockingQueue;

	import android.os.SystemClock;

	import com.android.im.engine.HeartbeatService;
	import com.android.im.engine.ImErrorInfo;
	import com.android.im.engine.ImException;
	import com.android.im.engine.SmsService;
	import com.android.im.engine.SystemService;
	import com.android.im.engine.SmsService.SmsListener;
	import com.android.im.engine.SmsService.SmsSendFailureCallback;

	public class SmsDataChannel extends DataChannel
	implements SmsListener, HeartbeatService.Callback {
	private SmsService mSmsService;
	private String mSmsAddr;
	private short mSmsPort;

	private long mLastActive;

	private SmsSplitter mSplitter;
	private SmsAssembler mAssembler;

	private LinkedBlockingQueue<Primitive> mReceiveQueue;

	private ImpsTransactionManager mTxManager;
	private boolean mConnected;
	private long mKeepAliveMillis;
	private Primitive mKeepAlivePrimitive;

	private long mReplyTimeout;
	private LinkedList<PendingTransaction> mPendingTransactions;
	private Timer mTimer;

	protected SmsDataChannel(ImpsConnection connection) throws ImException {
	super(connection);

	mTxManager = connection.getTransactionManager();

	ImpsConnectionConfig config = connection.getConfig();
	mReplyTimeout = config.getReplyTimeout();
	mSmsAddr = config.getSmsAddr();
	mSmsPort = (short) config.getSmsPort();
	mSmsService = SystemService.getDefault().getSmsService();

	mParser = new PtsPrimitiveParser();
	try {
	mSerializer = new PtsPrimitiveSerializer(config.getImpsVersion());
	} catch (SerializerException e) {
	throw new ImException(e);
	}
	mSplitter = new SmsSplitter(mSmsService.getMaxSmsLength());
	mAssembler = new SmsAssembler();
	mAssembler.setSmsListener(this);
	}

	@Override
	public void connect() throws ImException {
	mSmsService.addSmsListener(mSmsAddr, mSmsPort, mAssembler);
	mReceiveQueue = new LinkedBlockingQueue<Primitive>();
	mPendingTransactions = new LinkedList<PendingTransaction>();
	mTimer = new Timer(mReplyTimeout);
	new Thread(mTimer, "SmsDataChannel timer").start();
	mConnected = true;
	}

	@Override
	public long getLastActiveTime() {
	return mLastActive;
	}

	@Override
	public boolean isSendingQueueEmpty() {
	// Always true since we don't have a sending queue.
	return true;
	}

	@Override
	public Primitive receivePrimitive() throws InterruptedException {
	return mReceiveQueue.take();
	}

	@Override
	public void sendPrimitive(Primitive p) {
	ByteArrayOutputStream out = new ByteArrayOutputStream();
	try {
	mSerializer.serialize(p, out);
	mSplitter.split(out.toByteArray());
	SmsService smsService = SystemService.getDefault().getSmsService();
	SendFailureCallback sendFailureCallback
	= new SendFailureCallback(p.getTransactionID());
	while (mSplitter.hasNext()) {
	smsService.sendSms(mSmsAddr, mSmsPort, mSplitter.getNext(),
	sendFailureCallback);
	}
	mLastActive = SystemClock.elapsedRealtime();
	addPendingTransaction(p.getTransactionID());
	} catch (IOException e) {
	mTxManager.notifyErrorResponse(p.getTransactionID(),
	ImpsErrorInfo.SERIALIZER_ERROR, e.getLocalizedMessage());
	} catch (SerializerException e) {
	mTxManager.notifyErrorResponse(p.getTransactionID(),
	ImpsErrorInfo.SERIALIZER_ERROR, e.getLocalizedMessage());
	}
	}

	@Override
	public void shutdown() {
	mSmsService.removeSmsListener(this);
	mTimer.stop();
	mConnected = false;
	}

	@Override
	public void startKeepAlive(long interval) {
	if (!mConnected) {
	throw new IllegalStateException();
	}

	if (interval <= 0) {
	interval = mConnection.getConfig().getDefaultKeepAliveInterval();
	}

	mKeepAliveMillis = interval * 1000;
	if (mKeepAliveMillis < 0) {
	ImpsLog.log("Negative keep alive time. Won't send keep-alive");
	}
	mKeepAlivePrimitive = new Primitive(ImpsTags.KeepAlive_Request);

	HeartbeatService heartbeatService
	= SystemService.getDefault().getHeartbeatService();
	if (heartbeatService != null) {
	heartbeatService.startHeartbeat(this, mKeepAliveMillis);
	}
	}

	public long sendHeartbeat() {
	if (!mConnected) {
	return 0;
	}

	long inactiveTime = SystemClock.elapsedRealtime() - mLastActive;
	if (needSendKeepAlive(inactiveTime)) {
	sendKeepAlive();
	return mKeepAliveMillis;
	} else {
	return mKeepAliveMillis - inactiveTime;
	}
	}

	private void sendKeepAlive() {
	ImpsTransactionManager tm = mConnection.getTransactionManager();
	AsyncTransaction tx = new AsyncTransaction(tm) {
	@Override
	public void onResponseError(ImpsErrorInfo error) {
	}

	@Override
	public void onResponseOk(Primitive response) {
	// Since we never request a new timeout value, the response
	// can be ignored
	}
	};
	tx.sendRequest(mKeepAlivePrimitive);
	}

	private boolean needSendKeepAlive(long inactiveTime) {
	return mKeepAliveMillis - inactiveTime <= 500;
	}

	@Override
	public boolean resume() {
	return true;
	}

	@Override
	public void suspend() {
	// do nothing.
	}

	public void onIncomingSms(byte[] data) {
	try {
	Primitive p = mParser.parse(new ByteArrayInputStream(data));
	mReceiveQueue.put(p);
	removePendingTransaction(p.getTransactionID());
	} catch (ParserException e) {
	handleError(data, ImpsErrorInfo.PARSER_ERROR, e.getLocalizedMessage());
	} catch (IOException e) {
	handleError(data, ImpsErrorInfo.PARSER_ERROR, e.getLocalizedMessage());
	} catch (InterruptedException e) {
	handleError(data, ImpsErrorInfo.UNKNOWN_ERROR, e.getLocalizedMessage());
	}
	}

	private void handleError(byte[] data, int errCode, String info) {
	String trId = extractTrId(data);
	if (trId != null) {
	mTxManager.notifyErrorResponse(trId, errCode, info);
	removePendingTransaction(trId);
	}
	}

	private String extractTrId(byte[] data) {
	int transIdStart = 4;
	int index = transIdStart;
	while(Character.isDigit(data[index])) {
	index++;
	}
	int transIdLen = index - transIdStart;
	try {
	return new String(data, transIdStart, transIdLen, "UTF-8");
	} catch (UnsupportedEncodingException e) {
	return null;
	}
	}

	private void addPendingTransaction(String transId) {
	synchronized (mPendingTransactions) {
	mPendingTransactions.add(new PendingTransaction(transId));
	}
	}

	private void removePendingTransaction(String transId) {
	synchronized (mPendingTransactions) {
	Iterator<PendingTransaction> iter = mPendingTransactions.iterator();
	while (iter.hasNext()) {
	PendingTransaction tx = iter.next();
	if (tx.mTransId.equals(transId)) {
	iter.remove();
	break;
	}
	}
	}
	}

	/package/void checkTimeout() {
	synchronized (mPendingTransactions) {
	Iterator<PendingTransaction> iter = mPendingTransactions.iterator();
	while (iter.hasNext()) {
	PendingTransaction tx = iter.next();
	if (tx.isExpired(mReplyTimeout)) {
	notifyTimeout(tx);
	} else {
	break;
	}
	}
	}
	}

	private void notifyTimeout(PendingTransaction tx) {
	String transId = tx.mTransId;
	mTxManager.notifyErrorResponse(transId, ImpsErrorInfo.TIMEOUT,
	"Timeout");
	removePendingTransaction(transId);
	}

	private class SendFailureCallback implements SmsSendFailureCallback {
	private String mTransId;

	public SendFailureCallback(String transId) {
	mTransId = transId;
	}

	public void onFailure(int errorCode) {
	mTxManager.notifyErrorResponse(mTransId, ImErrorInfo.NETWORK_ERROR, null);
	}
	}

	private class Timer implements Runnable {
	private boolean mStopped;
	private long mInterval;

	public Timer(long interval) {
	mInterval = interval;
	mStopped = false;
	}

	public void stop() {
	mStopped = true;
	}

	public void run() {
	while (!mStopped) {
	try {
	Thread.sleep(mInterval);
	} catch (InterruptedException e) {
	continue;
	}
	checkTimeout();
	}
	}
	}

	private static class PendingTransaction {
	private String mTransId;
	private long mSentTime;

	public PendingTransaction(String transId) {
	mTransId = transId;
	}

	public boolean isExpired(long timeout) {
	return SystemClock.elapsedRealtime() - mSentTime >= timeout;
	}
	}
	}