cancel TileLoader job when thread shouldPause

- handle Exceptions in PbfTileDataSource, cleanup in finally{}

vtm/src/org/oscim/layers/tile/vector/VectorTileLoader.java

@@ -16,6 +16,8 @@
  */
 package org.oscim.layers.tile.vector;
 
+import java.util.concurrent.CancellationException;
+
 import org.oscim.core.GeometryBuffer.GeometryType;
 import org.oscim.core.MapElement;
 import org.oscim.core.MercatorProjection;
@@ -131,14 +133,18 @@ public class VectorTileLoader extends TileLoader implements IRenderTheme.Callbac
 
 		mTile = tile;
 		mTile.layers = new ElementLayers();
-
-		// query database, which calls 'process' callback
-		QueryResult result = mTileDataSource.executeQuery(mTile, this);
-
-		mTile = null;
-
-		clearState();
-
+		QueryResult result = null;
+		try {
+			// query database, which calls 'process' callback
+			result = mTileDataSource.executeQuery(mTile, this);
+		} catch (CancellationException e) {
+			log.debug("canceled {}", mTile);
+		} catch (Exception e) {
+			log.debug("{}", e);
+		} finally {
+			mTile = null;
+			clearState();
+		}
 		return (result == QueryResult.SUCCESS);
 	}
 
@@ -205,8 +211,12 @@ public class VectorTileLoader extends TileLoader implements IRenderTheme.Callbac
 
 	@Override
 	public void process(MapElement element) {
+
 		clearState();
 
+		if (isCanceled())
+			throw new CancellationException();
+
 		mElement = element;
 
 		if (element.type == GeometryType.POINT) {

vtm/src/org/oscim/tiling/source/common/LwHttp.java

@@ -56,6 +56,7 @@ public class LwHttp {
 	private Socket mSocket;
 	private OutputStream mCommandStream;
 	private InputStream mResponseStream;
+	private OutputStream mCacheOutputStream;
 	private long mLastRequest = 0;
 	private SocketAddress mSockAddr;
 
@@ -106,40 +107,34 @@ public class LwHttp {
 	}
 
 	static class Buffer extends BufferedInputStream {
-		final OutputStream mCache;
+		final OutputStream mCacheOutputstream;
 
 		public Buffer(InputStream is, OutputStream cache) {
 			super(is, 4096);
-			mCache = cache;
+			mCacheOutputstream = cache;
 		}
 
 		@Override
-		public synchronized int read() throws IOException {
+		public int read() throws IOException {
 			int data = super.read();
 			if (data >= 0)
-				mCache.write(data);
+				mCacheOutputstream.write(data);
 
 			return data;
 		}
 
 		@Override
-		public synchronized int read(byte[] buffer, int offset, int byteCount)
+		public int read(byte[] buffer, int offset, int byteCount)
 		        throws IOException {
 			int len = super.read(buffer, offset, byteCount);
 
 			if (len >= 0)
-				mCache.write(buffer, offset, len);
+				mCacheOutputstream.write(buffer, offset, len);
 
 			return len;
 		}
 	}
 
-	OutputStream mCacheOutputStream;
-
-	public void setOutputStream(OutputStream outputStream) {
-		mCacheOutputStream = outputStream;
-	}
-
 	public void close() {
 		if (mSocket != null) {
 			try {
@@ -177,6 +172,10 @@ public class LwHttp {
 			while (end < read && (buf[end] != '\n'))
 				end++;
 
+			if (end == BUFFER_SIZE) {
+				return null;
+			}
+
 			if (buf[end] != '\n')
 				continue;
 
@@ -209,7 +208,7 @@ public class LwHttp {
 
 			if (!ok) {
 				String line = new String(buf, pos, end - pos - 1);
-				log.debug(">" + line + "< ");
+				log.debug("> {} <", line);
 			}
 
 			pos += (end - pos) + 1;
@@ -255,12 +254,14 @@ public class LwHttp {
 			mMaxReq = RESPONSE_EXPECTED_LIVES;
 			// log.debug("create connection");
 		} else {
-			// FIXME not sure if this is correct way to drain socket
 			int avail = mResponseStream.available();
 			if (avail > 0) {
-				log.debug("Consume left-over bytes: " + avail);
-				while ((avail = mResponseStream.available()) > 0)
-					mResponseStream.read(buffer);
+				log.debug("left-over bytes: " + avail);
+				close();
+				lwHttpConnect();
+				// FIXME not sure if this is correct way to drain socket
+				//while ((avail = mResponseStream.available()) > 0)
+				//	mResponseStream.read(buffer);
 			}
 		}
 
@@ -353,8 +354,16 @@ public class LwHttp {
 		return true;
 	}
 
-	public void requestCompleted() {
+	public void setOutputStream(OutputStream outputStream) {
+		mCacheOutputStream = outputStream;
+	}
+
+	public void requestCompleted(boolean keepConnection) {
 		mLastRequest = System.nanoTime();
+		mCacheOutputStream = null;
+
+		if (!keepConnection)
+			close();
 	}
 
 	public int getContentLength() {

vtm/src/org/oscim/tiling/source/common/PbfTileDataSource.java

@@ -30,10 +30,6 @@ import org.oscim.utils.IOUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-/**
- *
- *
- */
 public abstract class PbfTileDataSource implements ITileDataSource {
 	static final Logger log = LoggerFactory.getLogger(PbfTileDataSource.class);
 
@@ -48,66 +44,55 @@ public abstract class PbfTileDataSource implements ITileDataSource {
 
 	@Override
 	public QueryResult executeQuery(MapTile tile, ITileDataSink sink) {
-		boolean success = true;
 
 		ITileCache.TileWriter cacheWriter = null;
 
 		if (mTileCache != null) {
 			ITileCache.TileReader c = mTileCache.getTile(tile);
 			if (c == null) {
-				// create new cache entry
 				cacheWriter = mTileCache.writeTile(tile);
-				mConn.setOutputStream(cacheWriter.getOutputStream());
 			} else {
+				InputStream is = c.getInputStream();
 				try {
-					InputStream is = c.getInputStream();
 					if (mTileDecoder.decode(tile, sink, is, c.getBytes())) {
-						IOUtils.closeQuietly(is);
 						return QueryResult.SUCCESS;
 					}
 				} catch (IOException e) {
 					e.printStackTrace();
+				} finally {
+					IOUtils.closeQuietly(is);
 				}
-				log.debug(tile + " Cache read failed");
 			}
 		}
 
+		boolean success = false;
 		try {
+			if (cacheWriter != null)
+				mConn.setOutputStream(cacheWriter.getOutputStream());
+
 			InputStream is;
 			if (!mConn.sendRequest(tile)) {
-				log.debug(tile + " Request failed");
-				success = false;
-			} else if ((is = mConn.readHeader()) != null) {
+				log.debug("{} Request failed", tile);
+			} else if ((is = mConn.readHeader()) == null) {
+				log.debug("{} Network Error", tile);
+			} else {
 				int bytes = mConn.getContentLength();
 				success = mTileDecoder.decode(tile, sink, is, bytes);
-				if (!success)
-					log.debug(tile + " Decoding failed");
-			} else {
-				log.debug(tile + " Network Error");
-				success = false;
 			}
 		} catch (SocketException e) {
-			log.debug(tile + " Socket exception: " + e.getMessage());
-			success = false;
+			log.debug("{} Socket exception: {}", tile, e.getMessage());
 		} catch (SocketTimeoutException e) {
-			log.debug(tile + " Socket Timeout");
-			success = false;
+			log.debug("{} Socket Timeout", tile);
 		} catch (UnknownHostException e) {
-			log.debug(tile + " No Network");
-			success = false;
-		} catch (Exception e) {
+			log.debug("{} No Network", tile);
+		} catch (IOException e) {
 			e.printStackTrace();
-			success = false;
+		} finally {
+			mConn.requestCompleted(success);
+
+			if (cacheWriter != null)
+				cacheWriter.complete(success);
 		}
-
-		mConn.requestCompleted();
-
-		if (cacheWriter != null)
-			cacheWriter.complete(success);
-
-		if (success)
-			mConn.close();
-
 		return success ? QueryResult.SUCCESS : QueryResult.FAILED;
 	}