Fetcher { public FileChunk get() throws IOException { if (useLocalFile) { LOG.info("Get pseudo fetch from local host"); startTime = System.currentTimeMillis(); finishTime = System.currentTimeMillis(); state = TajoProtos.FetcherState.FETCH_FINISHED; return fileChunk; } LOG.info("Get real fetch from remote host"); this.startTime = System.currentTimeMillis(); this.state = TajoProtos.FetcherState.FETCH_FETCHING; ChannelFuture future = null; try { future = bootstrap.clone().connect(new InetSocketAddress(host, port)) .addListener(ChannelFutureListener.CLOSE_ON_FAILURE); Channel channel = future.awaitUninterruptibly().channel(); if (!future.isSuccess()) { state = TajoProtos.FetcherState.FETCH_FAILED; throw new IOException(future.cause()); } String query = uri.getPath() + (uri.getRawQuery() != null ? "?" + uri.getRawQuery() : ""); HttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, query); request.headers().set(HttpHeaders.Names.HOST, host); request.headers().set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE); request.headers().set(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP); LOG.info("Status: " + getState() + ", URI:" + uri); channel.writeAndFlush(request); channel.closeFuture().syncUninterruptibly(); fileChunk.setLength(fileChunk.getFile().length()); return fileChunk; } finally { if(future != null && future.channel().isOpen()){ future.channel().close().awaitUninterruptibly(); } this.finishTime = System.currentTimeMillis(); LOG.info("Fetcher finished:" + (finishTime - startTime) + " ms, " + getState() + ", URI:" + uri); } } Fetcher(TajoConf conf, URI uri, FileChunk chunk); long getStartTime(); long getFinishTime(); long getFileLen(); TajoProtos.FetcherState getState(); int getMessageReceiveCount(); FileChunk get(); URI getURI();  }
@Test public void testGet() throws IOException { Random rnd = new Random(); QueryId queryId = QueryIdFactory.NULL_QUERY_ID; String sid = "1"; String partId = "1"; int partParentId = HashShuffleAppenderManager.getPartParentId(Integer.parseInt(partId), conf); String dataPath = conf.getVar(ConfVars.WORKER_TEMPORAL_DIR) + queryId.toString() + "/output/" + sid + "/hash-shuffle/" + partParentId + "/" + partId; String params = String.format("qid=%s&sid=%s&p=%s&type=%s", queryId, sid, partId, "h"); Path inputPath = new Path(dataPath); FSDataOutputStream stream = FileSystem.getLocal(conf).create(inputPath, true); for (int i = 0; i < 100; i++) { String data = ""+rnd.nextInt(); stream.write(data.getBytes()); } stream.flush(); stream.close(); URI uri = URI.create("http: FileChunk storeChunk = new FileChunk(new File(OUTPUT_DIR + "data"), 0, 0); storeChunk.setFromRemote(true); final Fetcher fetcher = new Fetcher(conf, uri, storeChunk); FileChunk chunk = fetcher.get(); assertNotNull(chunk); assertNotNull(chunk.getFile()); FileSystem fs = FileSystem.getLocal(new TajoConf()); FileStatus inStatus = fs.getFileStatus(inputPath); FileStatus outStatus = fs.getFileStatus(new Path(OUTPUT_DIR, "data")); assertEquals(inStatus.getLen(), outStatus.getLen()); assertEquals(TajoProtos.FetcherState.FETCH_FINISHED, fetcher.getState()); }