io.github.privacystreams.core

Class PStream

java.lang.Object

io.github.privacystreams.core.Stream

io.github.privacystreams.core.PStream

public class PStream
extends Stream

The interface of PStream (privacy stream). An PStream is a stream containing privacy sensitive items, and each item is an instance of `Item`. An PStream is produced by `uqi.getData` method. It can be transformed to another PStream using transformation functions, such as `filter`, `groupBy`, `map`, etc. Finally, it can be outputted using `asFieldList()`, `count`, etc.

Constructor Summary

Constructors

Constructor and Description
PStream(UQI uqi, Function<java.lang.Void,PStream> streamProvider)

Method Summary

Modifier and Type	Method and Description
java.util.List<Item>	asList() Collect the items in the stream to a list.
<TValue> java.util.List<TValue>	asList(java.lang.String fieldToSelect) Select a field in each item and output the field values to a list.
int	count() Count the number of items.
void	debug() Print the items for debugging.
PStream	filter(Function<Item,java.lang.Boolean> itemChecker) Filter the stream by testing an item with a function.
<TValue> PStream	filter(java.lang.String fieldName, TValue fieldValue) Filter the stream by checking whether a field equals a value.
void	forEach(Function<Item,java.lang.Void> callback) Callback with each item.
<TValue> void	forEach(java.lang.String fieldToSelect, Function<TValue,java.lang.Void> callback) Callback with a certain field of each item.
<TValue> TValue	getFieldAt(java.lang.String fieldName, int index) Get the N-th value of a given field.
Item	getFirst() Get the first item in the stream.
<TValue> TValue	getFirst(java.lang.String fieldName) Get the first value of the given field in the stream.
Item	getItemAt(int index) Pick the N-th item in the stream.
Function<java.lang.Void,PStream>	getStreamProvider()
<Tout> Function<java.lang.Void,Tout>	getValueGenerator(Function<PStream,Tout> streamOutputter) Get a value generator that can be evaluated on demand.
PStream	groupBy(java.lang.String groupField) Group the items according to a field.
void	idle() Do nothing with the items.
void	ifPresent(Function<Item,java.lang.Void> callback) Callback with an item once one item is present.
<TValue> void	ifPresent(java.lang.String fieldToSelect, Function<TValue,java.lang.Void> callback) Callback with a field value of an item once the field value is present.
PStream	inFixedInterval(long fixedInterval) Make the items be sent in a fixed interval.
PStream	keepChanges() Only keep the items that are different from the previous ones in the stream.
PStream	keepChanges(java.lang.String fieldName) Only Keep the items whose fields are different from the previous ones in the stream.
PStream	limit(Function<Item,java.lang.Boolean> itemChecker) Limit the stream by checking each item with a function.
PStream	limit(int maxCount) Limit the stream with a max number of items.
PStream	localGroupBy(java.lang.String groupField) Group the **contiguous** items according to a field.
PStream	logAs(java.lang.String logTag) Print the items in current stream.
PStream	logOverSocket(java.lang.String logTag) Print the items in current stream over socket.
PStream	map(Function<Item,Item> itemConverter) Convert each item in the stream with a function.
void	onChange(Function<Item,java.lang.Void> callback) Callback with an item when the item changes (is different from the previous item).
<TValue> void	onChange(java.lang.String fieldToSelect, Function<TValue,java.lang.Void> callback) Callback with a field value of an item when the field value changes.
<Tout> Tout	output(Function<java.util.List<Item>,Tout> itemsCollector) Output the items in the stream with a function.
<Tout> void	output(Function<java.util.List<Item>,Tout> resultComputer, Callback<Tout> resultHandler) Output the items in the stream with a function, and pass the result to a callback.
void	output(PStreamAction pStreamAction) Output the current PStream.
PStream	project(java.lang.String... fieldsToInclude) Project each item by including some fields.
PStream	reuse(int numOfReuses) Reuse current stream.
PStream	reverse() Reverse the order of items `reverse()` will reverse the order of the items in the stream.
PStream	sampleByCount(int stepCount) Sample the items based on a given step count.
PStream	sampleByInterval(long minInterval) Sample the items based on a given interval.
<TValue> PStream	setField(java.lang.String fieldToSet, Function<Item,TValue> fieldValueComputer) Set a field to a new value for each item in the stream.
<TValue> PStream	setGroupField(java.lang.String fieldToSet, Function<java.util.List<Item>,TValue> fieldValueComputer) Set a field to a new value for each item in the stream.
<TValue> PStream	setIndependentField(java.lang.String fieldToSet, Function<java.lang.Void,TValue> valueGenerator) Set the value of a new field with a value generator function.
PStream	shuffle() Shuffle the items.
PStream	sortBy(java.lang.String fieldName) Sort the items according to the value of a field, in ascending order.
PStream	timeout(long timeoutMilliseconds) Limit the stream with a timeout, stop the stream after time out.
PStream	transform(PStreamTransformation pStreamTransformation) Transform the current PStream to another PStream.
PStream	unGroup(java.lang.String unGroupField, java.lang.String newField) Un-group a list field in each item to multiple items.

Methods inherited from class io.github.privacystreams.core.Stream

getUQI, isClosed, register, toMap, toString, unregister, write

Methods inherited from class java.lang.Object

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

Constructor Detail

PStream

public PStream(UQI uqi,
               Function<java.lang.Void,PStream> streamProvider)

Method Detail

getStreamProvider

public Function<java.lang.Void,PStream> getStreamProvider()

Specified by:

getStreamProvider in class Stream

transform

public PStream transform(PStreamTransformation pStreamTransformation)

Transform the current PStream to another PStream.

Parameters:

pStreamTransformation - the function used to transform the stream

Returns:

the transformed stream

output

public void output(PStreamAction pStreamAction)

Output the current PStream.

Parameters:

pStreamAction - the function used to output stream

filter

@PSTransformation
public PStream filter(Function<Item,java.lang.Boolean> itemChecker)

Filter the stream by testing an item with a function. Specifically, keep the items that satisfy the function (aka. the function returns true). Eg. `filter(eq("x", 100))` will keep the items whose x field is equal to 100.

Parameters:

itemChecker - the function to check each item.

Returns:

The filtered stream.

filter

@PSTransformation
public <TValue> PStream filter(java.lang.String fieldName,
                                                 TValue fieldValue)

Filter the stream by checking whether a field equals a value. Specifically, keep the items in which the field equals the given value. Eg. `filter("x", 100)` will keep the items whose x field is equal to 100.

Parameters:

fieldName - the name of field to check

fieldValue - the value to compare with the field

Returns:

The filtered stream.

keepChanges

@PSTransformation
public PStream keepChanges()

Only keep the items that are different from the previous ones in the stream. Eg. a stream [1, 1, 2, 2, 2, 1, 1] will be [1, 2, 1] after `keepChanges()`

Returns:

the filtered stream.

keepChanges

@PSTransformation
public PStream keepChanges(java.lang.String fieldName)

Only Keep the items whose fields are different from the previous ones in the stream. Similar to `keepChanges()`, but only monitor a certain field

Parameters:

fieldName - the name of field to check whether an item should be kept

Returns:

the filtered stream.

sampleByInterval

@PSTransformation
public PStream sampleByInterval(long minInterval)

Sample the items based on a given interval. The items sent within the time interval since last item are dropped. Eg. If a stream has items sent at 1ms, 3ms, 7ms, 11ms and 40ms, `sampleByInterval(10)` will only keep the items sent at 1ms, 11ms and 40ms.

Parameters:

minInterval - the minimum interval (in milliseconds) between each two items.

Returns:

the filtered stream.

sampleByCount

@PSTransformation
public PStream sampleByCount(int stepCount)

Sample the items based on a given step count. The items are filtered to make sure `stepCount` number of items are dropped between each two new items. Eg. `sampleByCount(2)` will keep the 1st, 4th, 7th, 10th, ... items

Parameters:

stepCount - the num of items to drop since last item

Returns:

the filtered stream

limit

@PSTransformation
public PStream limit(Function<Item,java.lang.Boolean> itemChecker)

Limit the stream by checking each item with a function. Specifically, keep the stream as long as the checker holds (aka. the checker function returns true). Eg. `limit(eq("x", 100))` will keep all items in the stream as long as x field equals to 100, once an item's x value is not equal to 100, the stream stops.

Parameters:

itemChecker - the function to check each item.

Returns:

The limited stream.

limit

@PSTransformation
public PStream limit(int maxCount)

Limit the stream with a max number of items. Specifically, stop the stream if the count of items exceeds the threshold. Eg. `limit(10)` will limit the stream to at most 10 items

Parameters:

maxCount - the max number of items

Returns:

The limited stream.

timeout

@PSTransformation
public PStream timeout(long timeoutMilliseconds)

Limit the stream with a timeout, stop the stream after time out. Eg. `timeout(Duration.seconds(10))` will limit the stream to at most 10 seconds

Parameters:

timeoutMilliseconds - the timeout milliseconds

Returns:

The limited stream.

map

@PSTransformation
public PStream map(Function<Item,Item> itemConverter)

Convert each item in the stream with a function. Eg. `map(ItemOperators.setField("x", 10))` will set the "x" field of each item to 10 in the stream.

Parameters:

itemConverter - the function to map each item to another item

Returns:

The stream with items after mapping

inFixedInterval

@PSTransformation
public PStream inFixedInterval(long fixedInterval)

Make the items be sent in a fixed interval. Eg. If a stream has items sent at 1ms, 3ms, 7ms, 11ms and 40ms, `inFixedInterval(10)` will send items at 7ms, 11ms, 11ms and 40ms, in a 10ms interval.

Parameters:

fixedInterval - the fixed interval in milliseconds.

Returns:

The stream with items after mapping

project

@PSTransformation
public PStream project(java.lang.String... fieldsToInclude)

Project each item by including some fields. Other fields will be removed. Eg. `project("name", "email")` will only keep the "name" and "email" field in each item.

Parameters:

fieldsToInclude - the fields to include

Returns:

The stream with items after projection

setField

@PSTransformation
public <TValue> PStream setField(java.lang.String fieldToSet,
                                                   Function<Item,TValue> fieldValueComputer)

Set a field to a new value for each item in the stream. The value is computed with a function that take the item as input. Eg. `setField("x", Comparators.gt("y", 10))` will set a new boolean field "x" to each item, which indicates whether the "y" field is greater than 10.

Type Parameters:

TValue - the type of the new field value

Parameters:

fieldToSet - the name of the field to set.

fieldValueComputer - the function to compute the value of the new field based on each item.

Returns:

the stream of items with the new field set

setGroupField

@PSTransformation
public <TValue> PStream setGroupField(java.lang.String fieldToSet,
                                                        Function<java.util.List<Item>,TValue> fieldValueComputer)

Set a field to a new value for each item in the stream. This transformation can only be used after invoking group methods (`groupBy`, `localGroupBy`). The value is computed with a function that takes the grouped items as input at runtime. Eg. `setGroupField("count", StatisticOperators.count())` will set a new field "count" to each item, which represents the number of items in the grouped sub stream.

Type Parameters:

TValue - the type of the new field value

Parameters:

fieldToSet - the name of the field to set.

fieldValueComputer - the function to compute the new field value, which takes the list of grouped items as input.

Returns:

the stream of items with the new field set

setIndependentField

@PSTransformation
public <TValue> PStream setIndependentField(java.lang.String fieldToSet,
                                                              Function<java.lang.Void,TValue> valueGenerator)

Set the value of a new field with a value generator function. The value generator function is independent from current item, which does not need a input (input type is Void). The value generator will be evaluated on demand at runtime. Eg. `setIndependentField("time", TimeOperators.getCurrentTime())` will set the field "time" to a timestamp in each item; `setIndependentField("wifiStatus", DeviceOperators.isWifiConnected())` will set the field "wifiStatus" to a boolean indicating whether wifi is connected in each item.

Type Parameters:

TValue - the type of the new field value.

Parameters:

fieldToSet - the name of the field to set.

valueGenerator - the function to compute the field value.

Returns:

the stream of items with the new field set

sortBy

@PSTransformation(changeOrder=true)
public PStream sortBy(java.lang.String fieldName)

Sort the items according to the value of a field, in ascending order. Eg. `sortBy("timestamp")` will sort the items in the stream by timestamp field.

Parameters:

fieldName - the field used to sort the items in current stream, in ascending order

Returns:

The stream with sorted items.

shuffle

@PSTransformation(changeOrder=true)
public PStream shuffle()

Shuffle the items. `shuffle()` will randomize the order of the items in the stream.

Returns:

The stream with shuffled items.

reverse

@PSTransformation(changeOrder=true)
public PStream reverse()

Reverse the order of items `reverse()` will reverse the order of the items in the stream.

Returns:

The stream with reversed items.

groupBy

@PSTransformation(changeOrder=true)
public PStream groupBy(java.lang.String groupField)

Group the items according to a field. After grouping, the items in the new stream will only have two fields. One is the field used for grouping by. Another is "grouped_items" which is a list of grouped Items. Eg. `groupBy("x")` will group the items with same "x" field, and the item in the stream after groupping will contain two fields: "x" and "grouped_items".

Parameters:

groupField - the field used to group the items in current stream.

Returns:

The grouped stream

localGroupBy

@PSTransformation
public PStream localGroupBy(java.lang.String groupField)

Group the **contiguous** items according to a field. After grouping, the items in the new stream will only have two fields. One is the field used for grouping by. Another is "grouped_items" which is a list of grouped Items. Eg. `localGroupBy("x")` will group the contiguous items with same "x" field, and the item in the stream after groupping will contain two fields: "x" and "grouped_items".

Parameters:

groupField - the field used to group the items in current stream.

Returns:

The grouped stream

unGroup

@PSTransformation
public PStream unGroup(java.lang.String unGroupField,
                                         java.lang.String newField)

Un-group a list field in each item to multiple items. Each element in the list will be a new field in each item of the new stream. After un-grouping, the items in the new streams will have the same amount of fields as the original stream. However, the list field (`unGroupField`) will be replaced by a new field (`newField`). Eg. `unGroup("emails", "email")` will un-group the "emails" field (which is a list) in an item to several new items with a "email" field.

Parameters:

unGroupField - the field to un-group, whose value should be a list

newField - the new field name in the new stream

Returns:

The un-grouped stream

output

@PSAction(blocking=false)
public <Tout> void output(Function<java.util.List<Item>,Tout> resultComputer,
                                                    Callback<Tout> resultHandler)

Output the items in the stream with a function, and pass the result to a callback. This method will not be blocked. Eg. `outputItems(StatisticOperators.count(), new Callback(){...})` will count the number of items and callback with the number.

Type Parameters:

Tout - the type of the result

Parameters:

resultComputer - the function used to compute result based on the items in current stream

resultHandler - the function to handle the result

output

@PSAction(blocking=true)
public <Tout> Tout output(Function<java.util.List<Item>,Tout> itemsCollector)
                                            throws PSException

Output the items in the stream with a function. This method will block until the result returns. Eg. `output(StatisticOperators.count())` will output the number of items.

Type Parameters:

Tout - the type of the result

Parameters:

itemsCollector - the function used to output current stream

Returns:

the result

Throws:

PSException - if failed to the result.

getFirst

@PSAction(blocking=true)
public Item getFirst()
                                       throws PSException

Get the first item in the stream.

Returns:

the first item of current PStream

Throws:

PSException

getFirst

@PSAction(blocking=true)
public <TValue> TValue getFirst(java.lang.String fieldName)
                                                  throws PSException

Get the first value of the given field in the stream.

Returns:

the first item of current PStream

Throws:

PSException

getItemAt

@PSAction(blocking=true)
public Item getItemAt(int index)
                                        throws PSException

Pick the N-th item in the stream. N is the index.

Parameters:

index - the index of target item.

Returns:

the item selected from the current PStream

Throws:

PSException

getFieldAt

@PSAction(blocking=true)
public <TValue> TValue getFieldAt(java.lang.String fieldName,
                                                           int index)
                                                    throws PSException

Get the N-th value of a given field. N is the index.

Parameters:

fieldName - the name of the field to select

index - the index of target item.

Returns:

the item selected from the current PStream

Throws:

PSException

idle

@PSAction(blocking=false)
public void idle()

Do nothing with the items.

debug

@PSAction(blocking=false)
public void debug()

Print the items for debugging.

logAs

@PSTransformation
public PStream logAs(java.lang.String logTag)

Print the items in current stream.

Parameters:

logTag - the log tag to use in printing current stream

logOverSocket

@PSTransformation
public PStream logOverSocket(java.lang.String logTag)

Print the items in current stream over socket.

Parameters:

logTag - the log tag to use in printing current stream

count

@PSAction(blocking=true)
public int count()
                                   throws PSException

Count the number of items.

Returns:

the count of number of items in the stream.

Throws:

PSException

asList

@PSAction(blocking=true)
public java.util.List<Item> asList()
                                                     throws PSException

Collect the items in the stream to a list. Each item in the list is an instance of `Item`.

Returns:

a list of key-value maps, each map represents an item

Throws:

PSException

asList

@PSAction(blocking=true)
public <TValue> java.util.List<TValue> asList(java.lang.String fieldToSelect)
                                                                throws PSException

Select a field in each item and output the field values to a list.

Type Parameters:

TValue - the type of field value

Parameters:

fieldToSelect - the field to select

Returns:

a list of field values

Throws:

PSException

forEach

@PSAction(blocking=false)
public void forEach(Function<Item,java.lang.Void> callback)

Callback with each item.

Parameters:

callback - the callback to invoke for each item.

forEach

@PSAction(blocking=false)
public <TValue> void forEach(java.lang.String fieldToSelect,
                                                       Function<TValue,java.lang.Void> callback)

Callback with a certain field of each item.

Type Parameters:

TValue - the type of the field

Parameters:

fieldToSelect - the name of the field to callback with

callback - the callback to invoke for each item field

onChange

@PSAction(blocking=false)
public void onChange(Function<Item,java.lang.Void> callback)

Callback with an item when the item changes (is different from the previous item).

Parameters:

callback - the callback to invoke for the changed item

onChange

@PSAction(blocking=false)
public <TValue> void onChange(java.lang.String fieldToSelect,
                                                        Function<TValue,java.lang.Void> callback)

Callback with a field value of an item when the field value changes.

Type Parameters:

TValue - the type of the field

Parameters:

fieldToSelect - the name of the field to callback with

callback - the callback to invoke for the changed item field

ifPresent

@PSAction(blocking=false)
public void ifPresent(Function<Item,java.lang.Void> callback)

Callback with an item once one item is present.

Parameters:

callback - the callback to invoke once the item is present

ifPresent

@PSAction(blocking=false)
public <TValue> void ifPresent(java.lang.String fieldToSelect,
                                                         Function<TValue,java.lang.Void> callback)

Callback with a field value of an item once the field value is present.

Type Parameters:

TValue - the type of the field

Parameters:

fieldToSelect - the name of the field to callback with

callback - the callback to invoke once the field value is present

reuse

@PSTransformation
public PStream reuse(int numOfReuses)

Reuse current stream.

Overrides:

reuse in class Stream

Parameters:

numOfReuses - number of reuses

Returns:

the stream ready for reuse

getValueGenerator

public <Tout> Function<java.lang.Void,Tout> getValueGenerator(Function<PStream,Tout> streamOutputter)

Get a value generator that can be evaluated on demand. The generator will not be evaluated immediately, instead, it will be evaluated once `apply()` is called.

Parameters:

streamOutputter - the function to output the stream

Returns:

the value generator