Files
spark/src/DataStreamSuite/DataStream/Primitives/StaticCollectionStream/BlueTableStream.cs
T

61 lines
2.0 KiB
C#

using System;
using System.Collections;
using System.Reflection;
using BCG.Utilities;
namespace BCG {
// if a table needs to be fired, it can be wrapped around this
public class BlueTableStream : ViewStream {
private IBlueRowCollection _rows;
public BlueTableStream(IBlueRowCollection rows) {
_rows = rows;
}
protected override void myOnFirstSubscriber() {
var dp_out = BlueDataUtility.build_dp_inc(_rows, myLastValue as DataPacket);
myFireEvent(dp_out);
}
}
[DataStreamUserRespresentation(false)]
public class BlueRowCollectionStaticStream : ViewStream {
public BlueRowCollectionStaticStream() { }
public BlueRowCollectionStaticStream(IBlueRowCollection rows) { loadCollection(rows); }
public void loadCollection(IBlueRowCollection rows) {
myFireEvent(new DataPacket(rows, rows, myLastValue != null ?
new BlueTable((myLastValue as DataPacket).newRows) : new BlueTable(rows.schema)));
}
public void crush_with_single_row(IBlueRow r) {
BlueTable t = new BlueTable(r.schema);
t.addRow(r);
loadCollection(t);
}
public void loadPacket(DataPacket dp) { myFireEvent(dp); }
/// <summary>
/// Will work with streams that take ViewStream as first argument ex project, filter, distinct etc
/// The other arguments are provided by the user.
/// </summary>
/// <param name="rows"></param>
/// <param name="streamType"></param>
/// <param name="args"></param>
/// <returns></returns>
public static IBlueRowCollection statelessOp(IBlueRowCollection rows, Type streamType, params object[] args) {
ArrayList totArgs = new ArrayList(args);
BlueRowCollectionStaticStream v1 = new BlueRowCollectionStaticStream();
totArgs.Insert(0, v1);
ViewStream retVal = Activator.CreateInstance(streamType, totArgs.ToArray()) as ViewStream;
v1.loadCollection(rows);
return retVal.rows;
}
}
}