au.gov.amsa.ais.rx

Class Streams

java.lang.Object

au.gov.amsa.ais.rx.Streams

public class Streams
extends Object

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	Streams.TimestampedAndLine<T extends AisMessage>
static class	Streams.TimestampedAndLines<T extends AisMessage>

Field Summary

Fields

Modifier and Type	Field and Description
static int	BUFFER_SIZE
static rx.functions.Func1<String,com.google.common.base.Optional<NmeaMessage>>	LINE_TO_NMEA_MESSAGE
static rx.functions.Func1<NmeaMessage,com.google.common.base.Optional<Timestamped<AisMessage>>>	TO_AIS_MESSAGE
static rx.functions.Func1<NmeaMessage,Streams.TimestampedAndLine<AisMessage>>	TO_AIS_MESSAGE_AND_LINE
static rx.functions.Func1<com.google.common.base.Optional<List<NmeaMessage>>,Streams.TimestampedAndLines<AisMessage>>	TO_AIS_MESSAGE_AND_LINES

Constructor Summary

Constructors

Constructor and Description
Streams()

Method Summary

Modifier and Type	Method and Description
static rx.Observable.Transformer<NmeaMessage,com.google.common.base.Optional<List<NmeaMessage>>>	addToBuffer(int bufferSize)
static rx.Observable.Transformer<NmeaMessage,NmeaMessage>	aggregateMultiLineNmea(int bufferSize)
static rx.Observable.Transformer<NmeaMessage,NmeaMessage>	aggregateMultiLineNmeaWithLines(int bufferSize)
static rx.Observable<String>	connect(String host, int port)
static rx.Observable<Streams.TimestampedAndLine<AisMessage>>	connectAndExtract(String host, int port)
static rx.Observable<String>	connectOnce(HostPort hostPort)
static rx.Observable<Streams.TimestampedAndLine<AisMessage>>	extract(rx.Observable<String> rawAisNmea)
static rx.Observable<Fix>	extractFixes(rx.Observable<String> rawAisNmea)
static rx.functions.Func1<List<File>,rx.Observable<Integer>>	extractFixesFromNmeaGzAndAppendToFile(int linesPerProcessor, rx.Scheduler scheduler, rx.functions.Func1<Fix,String> fileMapper, int writeBufferSize, rx.functions.Action1<File> logger)
static rx.Observable<Timestamped<AisMessage>>	extractMessages(rx.Observable<String> rawAisNmea)
static rx.Observable<Streams.TimestampedAndLines<AisMessage>>	extractWithLines(rx.Observable<String> rawAisNmea)
static <T> rx.functions.Func1<com.google.common.base.Optional<T>,Boolean>	isPresent()
static void	main(String[] args)
static rx.Observable<String>	nmeaFrom(File file)
static rx.Observable<String>	nmeaFrom(InputStream is)
static rx.Observable<String>	nmeaFromGzip(File file)
static rx.Observable<String>	nmeaFromGzip(String filename)
static rx.Observable<rx.Observable<String>>	nmeasFromGzip(rx.Observable<File> files)
static void	print(rx.Observable<?> stream)
static void	print(rx.Observable<?> stream, PrintStream out)
static rx.functions.Func1<String,rx.Observable<LineAndTime>>	toLineAndTime()
static <T> rx.functions.Func1<com.google.common.base.Optional<T>,T>	toValue()
static <T> rx.Observable.Transformer<com.google.common.base.Optional<T>,T>	valueIfPresent()
static rx.Observable<Integer>	writeFixesFromNmeaGz(File input, Pattern inputPattern, File output, int logEvery, int writeBufferSize, rx.Scheduler scheduler, int linesPerProcessor, long downSampleIntervalMs, rx.functions.Func1<Fix,String> fileMapper)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

BUFFER_SIZE

public static final int BUFFER_SIZE

See Also:

Constant Field Values

LINE_TO_NMEA_MESSAGE

public static final rx.functions.Func1<String,com.google.common.base.Optional<NmeaMessage>> LINE_TO_NMEA_MESSAGE

TO_AIS_MESSAGE

public static final rx.functions.Func1<NmeaMessage,com.google.common.base.Optional<Timestamped<AisMessage>>> TO_AIS_MESSAGE

TO_AIS_MESSAGE_AND_LINE

public static final rx.functions.Func1<NmeaMessage,Streams.TimestampedAndLine<AisMessage>> TO_AIS_MESSAGE_AND_LINE

TO_AIS_MESSAGE_AND_LINES

public static final rx.functions.Func1<com.google.common.base.Optional<List<NmeaMessage>>,Streams.TimestampedAndLines<AisMessage>> TO_AIS_MESSAGE_AND_LINES

Constructor Detail

Streams

public Streams()

Method Detail

connect

public static rx.Observable<String> connect(String host,
                                            int port)

connectAndExtract

public static rx.Observable<Streams.TimestampedAndLine<AisMessage>> connectAndExtract(String host,
                                                                                      int port)

extract

public static rx.Observable<Streams.TimestampedAndLine<AisMessage>> extract(rx.Observable<String> rawAisNmea)

extractWithLines

public static rx.Observable<Streams.TimestampedAndLines<AisMessage>> extractWithLines(rx.Observable<String> rawAisNmea)

extractMessages

public static rx.Observable<Timestamped<AisMessage>> extractMessages(rx.Observable<String> rawAisNmea)

isPresent

public static <T> rx.functions.Func1<com.google.common.base.Optional<T>,Boolean> isPresent()

toValue

public static <T> rx.functions.Func1<com.google.common.base.Optional<T>,T> toValue()

valueIfPresent

public static <T> rx.Observable.Transformer<com.google.common.base.Optional<T>,T> valueIfPresent()

extractFixes

public static rx.Observable<Fix> extractFixes(rx.Observable<String> rawAisNmea)

nmeaFrom

public static rx.Observable<String> nmeaFrom(File file)

nmeaFrom

public static rx.Observable<String> nmeaFrom(InputStream is)

nmeaFromGzip

public static rx.Observable<String> nmeaFromGzip(String filename)

nmeasFromGzip

public static rx.Observable<rx.Observable<String>> nmeasFromGzip(rx.Observable<File> files)

nmeaFromGzip

public static rx.Observable<String> nmeaFromGzip(File file)

print

public static void print(rx.Observable<?> stream,
                         PrintStream out)

print

public static void print(rx.Observable<?> stream)

toLineAndTime

public static final rx.functions.Func1<String,rx.Observable<LineAndTime>> toLineAndTime()

addToBuffer

public static final rx.Observable.Transformer<NmeaMessage,com.google.common.base.Optional<List<NmeaMessage>>> addToBuffer(int bufferSize)

aggregateMultiLineNmea

public static final rx.Observable.Transformer<NmeaMessage,NmeaMessage> aggregateMultiLineNmea(int bufferSize)

aggregateMultiLineNmeaWithLines

public static final rx.Observable.Transformer<NmeaMessage,NmeaMessage> aggregateMultiLineNmeaWithLines(int bufferSize)

connectOnce

public static rx.Observable<String> connectOnce(HostPort hostPort)

extractFixesFromNmeaGzAndAppendToFile

public static rx.functions.Func1<List<File>,rx.Observable<Integer>> extractFixesFromNmeaGzAndAppendToFile(int linesPerProcessor,
                                                                                                          rx.Scheduler scheduler,
                                                                                                          rx.functions.Func1<Fix,String> fileMapper,
                                                                                                          int writeBufferSize,
                                                                                                          rx.functions.Action1<File> logger)

writeFixesFromNmeaGz

public static rx.Observable<Integer> writeFixesFromNmeaGz(File input,
                                                          Pattern inputPattern,
                                                          File output,
                                                          int logEvery,
                                                          int writeBufferSize,
                                                          rx.Scheduler scheduler,
                                                          int linesPerProcessor,
                                                          long downSampleIntervalMs,
                                                          rx.functions.Func1<Fix,String> fileMapper)

main

public static void main(String[] args)