Files
spark/src/DataStreamSuite/DataStream/Primitives/Project/ProjectStream.cs
T
2024-06-06 09:20:05 -05:00

102 lines
3.4 KiB
C#

namespace BCG {
using Spark.Core.Collections;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
[DataStreamUserRespresentation("oldproject")]
public class ProjectStream : ViewStream
{
private ProjectHelper myHelper;
private string myColExp, myTableName;
private ViewStream myVs;
private IDictionary<string, string> myDic;
/// <summary>
/// </summary>
/// <param name="vs">stream to susbsribe to as the input stream</param>
/// <param name="colExp">expression defining the column mapping</param>
public ProjectStream (ViewStream vs, string colExp)
{
myVs = vs;
myColExp = colExp;
}
/// <summary>
/// </summary>
/// <param name="vs">stream to subscribe to as the input stream</param>
/// <param name="colExp">expression defining the column mapping</param>
/// <param name="tableName">name of the mapped table</param>
public ProjectStream (ViewStream vs, string colExp, string tableName)
{
if (colExp == "!")
mySetRealSource (new ChangeTableNameStream (vs, tableName));
else
{
myVs = vs;
myColExp = colExp;
myTableName = tableName;
}
}
/// <summary>
/// </summary>
/// <param name="vs">stream to subscribe to as the input stream</param>
/// <param name="dic">dictionary mapping the aliased to the original</param>
public ProjectStream (ViewStream vs, IDictionary dic)
: this(vs, GenericDictionaryWrapper.Create<string, string>(dic)) {}
public ProjectStream (ViewStream vs, IEnumerable<KeyValuePair<string, string>> dic)
{
myVs = vs;
myDic = dic.ToDictionary(kv => kv.Value, kv => kv.Key);
}
protected override void myOnFirstSubscriber()
{
myVs.onChange += new BCG.DataStream.ChangeDelegate (myHandleChange);
}
protected override void myOnLastSubscriber()
{
myVs.onChange -= new BCG.DataStream.ChangeDelegate (myHandleChange);;
}
internal void myHandleChange (object newValue)
{
DataPacket dp = newValue as DataPacket;
if (myHelper != null)
{
IBlueRowCollection dels, adds;
BlueDataUtility.getTrueDeletesAndAdds (dp.deletedRows, dp.addedRows, out dels, out adds, myHelper.relevantOrigCols);
if (dels.Count + adds.Count > 0)
myFireEvent (new DataPacket (new ProjectedTable (dp.newRows, myHelper),
new ProjectedTable (adds, myHelper),
new ProjectedTable (dels, myHelper)
)
);
}
else
{ //first time a callback is received
myHelper = myDic == null ?
new ProjectHelper (dp.schema, myColExp, myTableName) :
new ProjectHelper (dp.schema, myDic, myTableName);
IBlueRowCollection temp;
myFireEvent (new DataPacket (temp = new ProjectedTable (dp.newRows, myHelper),
temp,
new BlueTable (temp.schema)
)
);
}
}
public void myHandleChangeTest(object newValue)
{
myHandleChange(newValue);
}
}
}