1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.commons.pipeline.stage;
19
20 import java.io.IOException;
21 import java.io.InputStream;
22 import java.net.MalformedURLException;
23 import java.net.URL;
24 import java.util.ArrayList;
25 import java.util.List;
26 import java.util.Queue;
27 import org.apache.commons.logging.Log;
28 import org.apache.commons.logging.LogFactory;
29 import org.apache.commons.pipeline.stage.BaseStage;
30 import org.apache.commons.pipeline.StageException;
31
32
33
34
35
36 public class URLToInputStreamStage extends BaseStage {
37 private final Log log = LogFactory.getLog(URLToInputStreamStage.class);
38 private List<InputStream> inputStreams = new ArrayList<InputStream>();
39
40
41 public URLToInputStreamStage() { }
42
43
44
45
46
47
48
49 public void process(Object obj) throws org.apache.commons.pipeline.StageException {
50 URL url = null;
51 if (obj instanceof URL){
52 url = (URL) obj;
53 } else if (obj instanceof String) {
54 String urlString = (String) obj;
55 try {
56 url = new URL(urlString);
57 } catch (MalformedURLException e){
58 throw new StageException(this, "Error converting url String:" + urlString, e);
59 }
60 }
61
62 try {
63 InputStream inputStream = url.openStream();
64 this.inputStreams.add(inputStream);
65 log.info("enqueing input stream");
66 this.emit(inputStream);
67 } catch (IOException e){
68 throw new StageException(this, "Error with stream from url:" + url, e);
69 }
70 }
71
72
73
74
75 public void release() {
76 log.info("running post process number of streams:" + inputStreams.size());
77 while(inputStreams.size() > 0){
78 InputStream is = (InputStream) inputStreams.remove(0);
79 try {
80 is.close();
81 log.info("closed stream");
82 } catch (IOException e){
83 log.warn("Error closing stream",e);
84 }
85 }
86 }
87 }