I'm implementing a download application that works like android's
DownloadProvider but based HttpCore NIO.
A fatal problem I encountered is this:
04-29 09:26:05.419: ERROR/AndroidRuntime(332):
java.nio.channels.IllegalSelectorException
04-29 09:26:05.419: ERROR/AndroidRuntime(332): at
java.nio.channels.spi.AbstractSelectableChannel.register(AbstractSelectableChannel.java:158)
04-29 09:26:05.419: ERROR/AndroidRuntime(332): at
java.nio.channels.SelectableChannel.register(SelectableChannel.java:158)
04-29 09:26:05.419: ERROR/AndroidRuntime(332): at
org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor.processSessionRequests(DefaultConnectingIOReactor.java:244)
04-29 09:26:05.419: ERROR/AndroidRuntime(332): at
org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor.processEvents(DefaultConnectingIOReactor.java:97)
04-29 09:26:05.419: ERROR/AndroidRuntime(332): at
org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor.execute(AbstractMultiworkerIOReactor.java:317)
04-29 09:26:05.419: ERROR/AndroidRuntime(332): at
io.DownloadClient.run(DownloadClient.java:168)
04-29 09:26:05.419: ERROR/AndroidRuntime(332): at
java.lang.Thread.run(Thread.java:1096)
however it can work on real j2se6, is that unsupported on android at all?
And a minor problem: Will "ioReactor.execute(ioEventDispatch)" really throw out
a InterruptedIOException?
I found it will be never even when I interrupt the thread explicitly. shutdown
can't be done gracefully without this.
Thanks for help, my raw codes for testing:
//********** io.DownloadClient.java **********
package io;
import java.io.File;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.http.HttpException;
import org.apache.http.HttpHost;
import org.apache.http.HttpRequest;
import org.apache.http.HttpResponse;
import org.apache.http.impl.DefaultConnectionReuseStrategy;
import org.apache.http.impl.nio.DefaultClientIOEventDispatch;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.nio.ContentDecoder;
import org.apache.http.nio.FileContentDecoder;
import org.apache.http.nio.NHttpClientConnection;
import org.apache.http.nio.NHttpClientHandler;
import org.apache.http.nio.entity.ConsumingNHttpEntity;
import org.apache.http.nio.protocol.AsyncNHttpClientHandler;
import org.apache.http.nio.protocol.NHttpRequestExecutionHandler;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOEventDispatch;
import org.apache.http.nio.util.HeapByteBufferAllocator;
import org.apache.http.params.BasicHttpParams;
import org.apache.http.params.CoreConnectionPNames;
import org.apache.http.params.CoreProtocolPNames;
import org.apache.http.params.HttpParams;
import org.apache.http.protocol.BasicHttpProcessor;
import org.apache.http.protocol.HttpContext;
import org.apache.http.protocol.RequestConnControl;
import org.apache.http.protocol.RequestContent;
import org.apache.http.protocol.RequestExpectContinue;
import org.apache.http.protocol.RequestTargetHost;
import org.apache.http.protocol.RequestUserAgent;
public class DownloadClient implements Runnable {
public static final String TAG = "nioDownload";
Thread dispatcher;
ExecutorService workers;
ConnectingIOReactor reactor;
IOEventDispatch eventDispatch;
public DownloadClient() throws IOException {
BasicHttpProcessor httpproc = new BasicHttpProcessor();
httpproc.addInterceptor(new RequestContent());
httpproc.addInterceptor(new RequestTargetHost());
httpproc.addInterceptor(new RequestConnControl());
httpproc.addInterceptor(new RequestUserAgent());
httpproc.addInterceptor(new RequestExpectContinue());
HttpParams params = new BasicHttpParams();
params.setParameter(CoreProtocolPNames.USER_AGENT,
"HttpComponents/1.1");
params.setIntParameter(CoreConnectionPNames.SO_TIMEOUT, 5000);
params.setIntParameter(CoreConnectionPNames.CONNECTION_TIMEOUT, 10000);
params.setIntParameter(CoreConnectionPNames.SOCKET_BUFFER_SIZE, 8 * 1024);
params.setBooleanParameter(CoreConnectionPNames.STALE_CONNECTION_CHECK,
false);
params.setBooleanParameter(CoreConnectionPNames.TCP_NODELAY, true);
NHttpRequestExecutionHandler execHandler = new NHttpRequestExecutionHandler()
{
HttpRequest request = null;
@Override
public void initalizeContext(HttpContext context, Object attachment) {
if(attachment != null && attachment instanceof DownloadCourse) {
DownloadCourse course = (DownloadCourse)attachment;
request = course.getRequest();
context.setAttribute(TAG, attachment);
}
}
@Override
public HttpRequest submitRequest(HttpContext context) {
HttpRequest result = this.request;
if(result != null) {
this.request = null;
}
return result;
}
@Override
public ConsumingNHttpEntity responseEntity(HttpResponse response,
HttpContext context)
throws IOException {
//nothing, handle in NHttpClientHandler.inputReady
return null;
}
@Override
public void handleResponse(HttpResponse response, HttpContext context)
throws IOException {
//nothing, handle in NHttpClientHandler.inputReady
}
@Override
public void finalizeContext(HttpContext context) {
context.removeAttribute(TAG);
}
};
NHttpClientHandler handler = new AsyncNHttpClientHandler(
httpproc, execHandler, new DefaultConnectionReuseStrategy(),
new HeapByteBufferAllocator(), params) {
@Override
public void closed(NHttpClientConnection conn) {
super.closed(conn);
Object attachment = conn.getContext().getAttribute(TAG);
if(attachment != null && attachment instanceof DownloadCourse) {
DownloadCourse course = (DownloadCourse)attachment;
course.interrupt();
}
}
@Override
public void exception(NHttpClientConnection conn, HttpException ex) {
super.exception(conn, ex);
Object attachment = conn.getContext().getAttribute(TAG);
if(attachment != null && attachment instanceof DownloadCourse) {
DownloadCourse course = (DownloadCourse)attachment;
course.error(ex);
}
}
@Override
public void exception(NHttpClientConnection conn, IOException ex) {
super.exception(conn, ex);
Object attachment = conn.getContext().getAttribute(TAG);
if(attachment != null && attachment instanceof DownloadCourse) {
DownloadCourse course = (DownloadCourse)attachment;
course.error(ex);
}
}
@Override
public void inputReady(NHttpClientConnection conn, ContentDecoder content) {
//don't invoke super, handle by self
log("inputReady: %s", content);
Object attachment = conn.getContext().getAttribute(TAG);
if(attachment != null && attachment instanceof DownloadCourse) {
DownloadCourse course = (DownloadCourse)attachment;
course.update((FileContentDecoder)content);
if(content.isCompleted()) {
course.complete(conn.getHttpResponse());
}
}
}
@Override
public void responseReceived(NHttpClientConnection conn) {
super.responseReceived(conn);
Object attachment = conn.getContext().getAttribute(TAG);
if(attachment != null && attachment instanceof DownloadCourse) {
DownloadCourse course = (DownloadCourse)attachment;
course.prepare(conn.getHttpResponse());
}
}
@Override
public void timeout(NHttpClientConnection conn) {
super.timeout(conn);
Object attachment = conn.getContext().getAttribute(TAG);
if(attachment != null && attachment instanceof DownloadCourse) {
DownloadCourse course = (DownloadCourse)attachment;
course.interrupt();
}
}
};
eventDispatch = new DefaultClientIOEventDispatch(handler, params);
reactor = new DefaultConnectingIOReactor(1, params);
}
@Override
public void run() {
log("DownloadClient run at %s", new Date());
try {
reactor.execute(eventDispatch);
} catch(InterruptedIOException e) {
//never thrown? even when being interrupted?
log("DownloadClient interrupted");
} catch(IOException e) {
e.printStackTrace();
} finally {
try {
reactor.shutdown();
} catch(IOException e) {
e.printStackTrace();
}
}
}
public void execute(DownloadCourse course) {
HttpHost host = course.getHost();
reactor.connect(
new InetSocketAddress(host.getHostName(), host.getPort() > 0 ?
host.getPort() : 80), null,
course, null);
}
public void startup() {
dispatcher = new Thread(this, "DownloadClient Thread");
dispatcher.setDaemon(true);
dispatcher.start();
workers = Executors.newFixedThreadPool(2);
}
public void shutdown() {
dispatcher.interrupt();
workers.shutdown();
}
public void download(URI uri, File saveFile) throws IOException {
DownloadCourse course = new DownloadCourse(this, uri, saveFile);
workers.submit(course);
}
public static void log(Object s) {
System.out.println(s);
}
public static void log(String format, Object... a) {
System.out.println(String.format(format, a));
}
public static void main(String[] args) throws Exception {
//startup client
DownloadClient client = new DownloadClient();
client.startup();
//submit request
File saveFile = new File("/sdcard/download.txt");
if(!saveFile.exists()) saveFile.createNewFile();
client.download(new URI("http://www.google.com/";), saveFile);
//for a while...
Thread.sleep(10000);
//clear for end
client.shutdown();
}
}
//********** io.DownloadCourse.java **********
package io;
import static io.DownloadClient.log;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.URI;
import java.nio.channels.FileChannel;
import java.util.Date;
import org.apache.commons.io.FileUtils;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.HttpRequest;
import org.apache.http.HttpResponse;
import org.apache.http.message.BasicHttpRequest;
import org.apache.http.nio.FileContentDecoder;
public class DownloadCourse implements Runnable {
public static final int maxRetryTimes = 5;
public static final byte QUEUING_STATUS = 0;
public static final byte DOWNLOADING = 1;
public static final byte PAUSED = 15;
public static final byte FINISHED = 127;
public static final byte VIOLATED = -1;
public static final byte TERMINATED = -128;
DownloadClient client;
URI uri;
File file;
RandomAccessFile randomAccessFile;
FileChannel fileChannel;
long contentLength;
long position;
int status;
int retryTimes;
public DownloadCourse(DownloadClient client, URI uri, File file) {
this.client = client;
this.uri = uri;
this.file = file;
status = QUEUING_STATUS;
}
@Override
public void run() {
log("DownloadCourse run at %s", new Date());
status = QUEUING_STATUS;
client.execute(this);
}
public HttpHost getHost() {
return uri.getPort() > 0 ?
new HttpHost(uri.getHost(), uri.getPort()) :
new HttpHost(uri.getHost());
}
public HttpRequest getRequest() {
HttpRequest request = uri.getQuery() != null ?
new BasicHttpRequest("GET", uri.getPath() + "?" + uri.getQuery()) :
new BasicHttpRequest("GET", uri.getPath());
request.setHeader("Host", getHost().toHostString());
return request;
}
public void prepare(HttpResponse response) {
Header contentLengthHeader = response.getFirstHeader("Content-Length");
if(contentLengthHeader != null) {
contentLength = Integer.parseInt(contentLengthHeader.getValue().trim());
}
try {
randomAccessFile = new RandomAccessFile(file, "rw");
if(contentLength > 0) {
randomAccessFile.setLength(contentLength);
}
fileChannel = randomAccessFile.getChannel();
} catch(IOException e) {
throw new RuntimeException(e);
}
status = DOWNLOADING;
}
public void update(FileContentDecoder content) {
long length = 0;
try {
length = content.transfer(fileChannel, position, contentLength);
} catch(IOException e) {
throw new RuntimeException(e);
}
position += length;
log("download progress: %d/%d", position, contentLength);
}
public void complete(HttpResponse response) {
try {
randomAccessFile.close();
log(response.getStatusLine().toString());
log(new String(FileUtils.readFileToByteArray(file)));
} catch(IOException e) {
throw new RuntimeException(e);
}
status = FINISHED;
}
public void error(Exception ex) {
try {
randomAccessFile.close();
} catch(IOException e) {
throw new RuntimeException(e);
}
if(++retryTimes < maxRetryTimes) {
status = VIOLATED;
} else {
status = TERMINATED;
}
}
public void interrupt() {
try {
randomAccessFile.close();
} catch(IOException e) {
throw new RuntimeException(e);
}
status = PAUSED;
}
public String toString() {
return String.format("%s\n\turi: %s\n\tfile: %s\n\tstatus:
%d\n\tcontentLength: %d\n\tposition:%d", super.toString(), uri, file, status,
contentLength, position);
}
}
