@@ -92,10 +92,10 @@ public class AnalyzeEmail { private static class SplitEmails extends SubAssembly { public SplitEmails(FetchPipe fetchPipe) { - Pipe splitPipe = new Pipe(SPLITTER_PIPE_NAME, fetchPipe.getTailPipe(FetchPipe.FETCHED_PIPE_NAME)); + Pipe splitPipe = new Pipe(SPLITTER_PIPE_NAME, fetchPipe.getContentTailPipe()); splitPipe = new Each(splitPipe, new MboxSplitterFunction()); // TODO KKr - code currently relies on splitPipe being first tail pipe. - setTails(splitPipe, fetchPipe.getTailPipe(FetchPipe.STATUS_PIPE_NAME)); + setTails(splitPipe, fetchPipe.getStatusTailPipe()); } } @@ -172,18 +172,17 @@ public class AnalyzeEmail { SimpleGroupingKeyGenerator grouping = new SimpleGroupingKeyGenerator(userAgent); IScoreGenerator scorer = new IScoreGenerator() { - public double generateScore(GroupedUrlDatum arg0) throws IOException { return 1.0; } + public double generateScore(GroupedUrlDatum arg0) { return 1.0; } }; IHttpFetcher fetcher = new SimpleHttpFetcher(MAX_THREADS, userAgent); FetchPipe fetchPagePipe = new FetchPipe(importPipe, grouping, scorer, fetcher); // Here's the pipe that will output UrlDatum tuples, by extracting URLs from the mod_mbox-generated page. - Pipe mboxPagePipe = new Each(fetchPagePipe.getTailPipe(FetchPipe.FETCHED_PIPE_NAME), - new ParseModMboxPageFunction(new Fields()), Fields.RESULTS); + Pipe mboxPagePipe = new Each(fetchPagePipe.getContentTailPipe(), new ParseModMboxPageFunction(new Fields()), Fields.RESULTS); // Create a named pipe for the status of the mod_mbox-generated pages. - Pipe mboxPageStatusPipe = new Pipe(MBOX_PAGE_STATUS_PIPE_NAME, fetchPagePipe.getTailPipe(FetchPipe.STATUS_PIPE_NAME)); + Pipe mboxPageStatusPipe = new Pipe(MBOX_PAGE_STATUS_PIPE_NAME, fetchPagePipe.getStatusTailPipe()); // Set up appropriate FetcherPolicy, where we increase the max content size (since mailbox files // can be big, e.g. 4MB). contrib/helpful/src/main/java/com/transpac/helpful/tools/AnalyzeEmail.java @@ -36,10 +36,10 @@ import cascading.tuple.Tuple; @SuppressWarnings("serial") public class FetchPipe extends SubAssembly { // Pipe that outputs FetchedDatum tuples, for URLs that were fetched. - public static final String FETCHED_PIPE_NAME = "fetched"; + public static final String CONTENT_PIPE_NAME = "FetchPipe-content"; // Pipe that outputs StatusDatum tuples, for all URLs being processed. - public static final String STATUS_PIPE_NAME = "status"; + public static final String STATUS_PIPE_NAME = "FetchPipe-status"; @SuppressWarnings({ "unchecked" }) private static class FilterErrorsFunction extends BaseOperation implements Function { @@ -83,7 +83,7 @@ public class FetchPipe extends SubAssembly { public MakeStatusFunction(Fields metaDataFields) { super(StatusDatum.FIELDS.append(metaDataFields)); - // Location of extra field added during fetch, that contains fetch error + // Location of extra field added during fetch, that contains fetch status _fieldPos = FetchedDatum.FIELDS.size() + metaDataFields.size(); _metaDataFields = metaDataFields; @@ -93,6 +93,9 @@ public class FetchPipe extends SubAssembly { public void operate(FlowProcess process, FunctionCall funcCall) { Tuple t = funcCall.getArguments().getTuple(); FetchedDatum fd = new FetchedDatum(t, _metaDataFields); + + // Get the fetch status that we hang on the end of the tuple, + // after all of the FetchedDatum fields. Comparable result = t.get(_fieldPos); StatusDatum status; @@ -140,13 +143,21 @@ public class FetchPipe extends SubAssembly { fetch = new Every(fetch, new FetcherBuffer(metaDataFields, fetcher), Fields.RESULTS); Fields fetchedFields = FetchedDatum.FIELDS.append(metaDataFields); - Pipe fetched = new Pipe(FETCHED_PIPE_NAME, new Each(fetch, new FilterErrorsFunction(fetchedFields))); + Pipe fetched = new Pipe(CONTENT_PIPE_NAME, new Each(fetch, new FilterErrorsFunction(fetchedFields))); Pipe status = new Pipe(STATUS_PIPE_NAME, new Each(fetch, new MakeStatusFunction(metaDataFields))); setTails(fetched, status); } - public Pipe getTailPipe(String pipeName) { + public Pipe getContentTailPipe() { + return getTailPipe(CONTENT_PIPE_NAME); + } + + public Pipe getStatusTailPipe() { + return getTailPipe(STATUS_PIPE_NAME); + } + + private Pipe getTailPipe(String pipeName) { String[] pipeNames = getTailNames(); for (int i = 0; i < pipeNames.length; i++) { if (pipeName.equals(pipeNames[i])) { @@ -169,7 +180,7 @@ public class FetchPipe extends SubAssembly { } result.put(STATUS_PIPE_NAME, statusSink); - result.put(FETCHED_PIPE_NAME, fetchedSink); + result.put(CONTENT_PIPE_NAME, fetchedSink); return result; } src/main/java/bixo/pipes/FetchPipe.java @@ -71,7 +71,7 @@ public class IndexingMetaDataTest { FetchPipe fetchPipe = new FetchPipe(pipe, grouping, scoring, fetcher, metaDataField); - ParsePipe parserPipe = new ParsePipe(fetchPipe.getTailPipe(FetchPipe.FETCHED_PIPE_NAME), new FakeParser(), metaDataField); + ParsePipe parserPipe = new ParsePipe(fetchPipe.getContentTailPipe(), new FakeParser(), metaDataField); Fields indexedFields = new Fields("text", "metaData"); Pipe indexPipe = new Each(parserPipe, new Fields(ParsedDatum.PARSED_TEXT_FIELD, "metaData"), new Identity(indexedFields)); src/test/java/bixo/pipes/IndexingMetaDataTest.java 2f8337896ad9dc5fbca7144fb621313ad5754b35 Ken Krugler kkrugler@transpac.com http://github.com/emi/bixo/commit/af0010b49c4370c06e7454bdfc1889311bdefaa7 af0010b49c4370c06e7454bdfc1889311bdefaa7 2009-10-09T12:40:25-07:00 2009-10-09T12:40:25-07:00 Clean up name of FetchPipe content output tail pipe. Added specific routines to get content & status tail pipes. Sync contrib/helpful with latest Bixo API. 093cb96aae79ec1e738ddc81864813b2ea784bb6 Ken Krugler kkrugler@transpac.com