im/imps/SmsDataChannel.java

/*
 * 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;
        }
    }
}