John Colvin
7d6e4efa4f
Nearly all warnings were due to nullability issues, since the concept of nullable reference types did not exist when the DataStream code was originally written. * Compression related files - BZip2InputStream (disable nullability warnings) - BZip2OutputStream (disable nullability warnings) - DeflaterHuffman (disable nullability warnings) - Inflater (disable nullability warnings) - InflaterDynHeader (disable nullability warnings) - ParallelJoinTest (disable nullability warnings) - TarHeader - TarInputStream (disable nullability warnings) - ZipEntry - ZipFile - ZipInputStream * AsyncTCPConnection_APM * AsyncTCPConnection_APM * InflaterHuffmanTree * ManagedSubscriber * ManagedTCPMessageHub * MapStream * PubSubClientMessage - Normalize indentation/whitespace - Fix some typos - Address nullability issues to fix build warnings - Modernize code * PubSubMessageReactor * PubSubServerManagedTCPConnection * PubSubSingleTagStateEvent * PubSubState * PublisherCore * PubsubDeserializationController * RemoteMessage - make 'message' nullable to resolve warnings elsewhere (in calling code) * SharedStreamConfig * SingleTagData * SubscriberCore * SuperMapStream * SystemHeartbeatPublisher * SystemStatsPublisher * SystemSubscriptionMapPublisher * ThreadSafeQueue * TradeBarsStream Only use Registry on windows.
370 lines
12 KiB
C#
370 lines
12 KiB
C#
#nullable disable
|
|
using System;
|
|
using System.Collections;
|
|
using System.Collections.Generic;
|
|
|
|
namespace BCG
|
|
{
|
|
|
|
/// <summary>
|
|
/// This class is used to effect a mapping between values in a join column and entries in three different tables:
|
|
/// the left and right tables of a join, as well as the output table (the join table).
|
|
/// </summary>
|
|
class ParallelJoinTestTablePointer
|
|
{
|
|
int myLeftTableId = -1;
|
|
public int LeftTableId
|
|
{
|
|
get { return myLeftTableId; }
|
|
set { myLeftTableId = value; }
|
|
}
|
|
|
|
int myRightTableId = -1;
|
|
public int RightTableId
|
|
{
|
|
get { return myRightTableId; }
|
|
set { myRightTableId = value; }
|
|
}
|
|
int myJoinTableId = -1;
|
|
public int JoinTableId
|
|
{
|
|
get { return myJoinTableId; }
|
|
set { myJoinTableId = value; }
|
|
}
|
|
|
|
public ParallelJoinTestTablePointer(int leftId, int rightId, int joinId)
|
|
{
|
|
myLeftTableId = leftId;
|
|
myRightTableId = rightId;
|
|
myJoinTableId = joinId;
|
|
}
|
|
public ParallelJoinTestTablePointer() : this(-1,-1,-1)
|
|
{
|
|
|
|
}
|
|
}
|
|
|
|
|
|
#region IBlueRow implementation for join and leftJoin
|
|
|
|
public class ParallelJoinTestRow : IBlueRow
|
|
{
|
|
private JoinProjectHelper myHelper;
|
|
private IBlueRow myLeftRow, myRightRow;
|
|
int myID;
|
|
|
|
public ParallelJoinTestRow(int id, IBlueRow leftRow, IBlueRow rightRow, JoinProjectHelper joinProjectHelper)
|
|
{
|
|
myID = id;
|
|
myLeftRow = leftRow;
|
|
myRightRow = rightRow;
|
|
myHelper = joinProjectHelper;
|
|
}
|
|
|
|
public int leftID { get { return (int)myLeftRow["id"]; } }
|
|
|
|
public int rightID { get { return (int)myRightRow["id"]; } }
|
|
|
|
public IBlueRow leftRow { get { return myLeftRow; } }
|
|
|
|
public IBlueRow rightRow { get { return myRightRow; } }
|
|
|
|
#region IBlueRow Members
|
|
|
|
public IBlueTableSchema schema { get { return myHelper.schema; } }
|
|
|
|
public object this[int index] { get { return this[schema[index].colName]; } }
|
|
|
|
public object this[string colName]
|
|
{
|
|
get
|
|
{
|
|
if (colName == "id")
|
|
return myID;
|
|
else
|
|
{
|
|
bool isLeft;
|
|
string col = myHelper.getOriginalCol(colName, out isLeft);
|
|
return isLeft ? myLeftRow[col] : (myRightRow == null ? (object)DBNull.Value : myRightRow[col]);
|
|
}
|
|
}
|
|
}
|
|
|
|
#endregion
|
|
|
|
#region IDictionary Members
|
|
|
|
public void Add(object key, object value) { { throw new NotSupportedException("IBlueRow is readonly"); } }
|
|
|
|
public void Clear() { throw new NotSupportedException("IBlueRow is readonly"); }
|
|
|
|
public bool Contains(object key) { throw new Exception("The method or operation is not implemented."); }
|
|
|
|
public System.Collections.IDictionaryEnumerator GetEnumerator() { return new BlueRowDictionaryEnumerator(this); }
|
|
|
|
public bool IsFixedSize { get { return true; } }
|
|
|
|
public bool IsReadOnly {get { return true; }}
|
|
|
|
public System.Collections.ICollection Keys { get { return schema.columnNames; } }
|
|
|
|
public void Remove(object key) { throw new NotSupportedException("IBlueRow is readonly"); }
|
|
|
|
public System.Collections.ICollection Values
|
|
{
|
|
get
|
|
{
|
|
List<object> rv = new List<object> ();
|
|
foreach (string col in Keys)
|
|
rv.Add(this[col]);
|
|
return rv;
|
|
}
|
|
}
|
|
|
|
public object this[object key]
|
|
{
|
|
get { return key is string ? this[key.ToString()] : this[(int)key]; }
|
|
set { throw new NotSupportedException("IBlueRow is readonly"); }
|
|
}
|
|
|
|
#endregion
|
|
|
|
#region ICollection Members
|
|
|
|
public void CopyTo(Array array, int index) { Values.CopyTo(array, index); }
|
|
|
|
public int Count { get { return schema.columnNames.Length; } }
|
|
|
|
public bool IsSynchronized { get { return false; } }
|
|
|
|
public object SyncRoot { get { return this; } }
|
|
|
|
#endregion
|
|
|
|
#region IEnumerable Members
|
|
|
|
System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator() { return new BlueRowEnumerator(this); }
|
|
|
|
#endregion
|
|
}
|
|
|
|
#endregion
|
|
/// <summary>
|
|
/// This class represents a join on two streams. It works by storing a table with the result of the join.
|
|
/// </summary>
|
|
public class ParallelJoinTestStream : ViewStream
|
|
{
|
|
BlueTable myJoinedTable;
|
|
BlueTable myLeftTable;
|
|
BlueTable myRightTable;
|
|
Hashtable myKeyToTablePointerMap;
|
|
ViewStream myVLeft, myVRight;
|
|
|
|
int myIdToUse;
|
|
string myLeftKey, myRightKey, myLeftProj, myRightProj;
|
|
public ParallelJoinTestStream(ViewStream leftStream, string leftKey, ViewStream rightStream, string rightKey)
|
|
: this(leftStream, leftKey, "!", rightStream, rightKey, "!") { }
|
|
public ParallelJoinTestStream(ViewStream leftStream, string leftKey, string leftProj, ViewStream rightStream, string rightKey, string rightProj)
|
|
{
|
|
myVLeft = leftStream;
|
|
myVRight = rightStream;
|
|
|
|
myLeftProj = leftProj;
|
|
myRightProj = rightProj;
|
|
|
|
myLeftKey = leftKey;
|
|
myRightKey = rightKey;
|
|
|
|
myKeyToTablePointerMap = new Hashtable();
|
|
myIdToUse = 0;
|
|
}
|
|
|
|
protected override void myOnFirstSubscriber()
|
|
{
|
|
myVLeft.onChange += new ChangeDelegate(myVLeft_onChange);
|
|
myVRight.onChange += new ChangeDelegate(myVRight_onChange);
|
|
}
|
|
|
|
void myVRight_onChange(object newValue)
|
|
{
|
|
myHandleChange(newValue as DataPacket, false);
|
|
|
|
}
|
|
|
|
void myVLeft_onChange(object newValue)
|
|
{
|
|
myHandleChange(newValue as DataPacket, true);
|
|
|
|
}
|
|
|
|
void myHandleChange(DataPacket dp, bool isLeft)
|
|
{
|
|
|
|
int i = 0; // for going throuhg loops and such
|
|
BlueTable addedRows = null, droppedRows = null;
|
|
|
|
if (myJoinedTable == null)
|
|
{
|
|
if (isLeft && myLeftTable == null)
|
|
{
|
|
myLeftTable = new BlueTable(dp.addedRows);
|
|
|
|
//go through the list of added rows and get the id of each row, storing this in the TablePointer map
|
|
foreach (IBlueRow row in myLeftTable)
|
|
{
|
|
if (myKeyToTablePointerMap.ContainsKey(row[myLeftKey]))
|
|
{
|
|
((ParallelJoinTestTablePointer)myKeyToTablePointerMap[row[myLeftKey]]).LeftTableId = (int)row["id"];
|
|
}
|
|
else
|
|
{
|
|
myKeyToTablePointerMap[row[myLeftKey]] = new ParallelJoinTestTablePointer((int)row["id"],-1,-1);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
else if (!isLeft && myRightTable == null)
|
|
{
|
|
myRightTable = new BlueTable(dp.addedRows);
|
|
//go through the list of added rows and get the id of each row, storing this in the TablePointer map
|
|
foreach (IBlueRow row in myRightTable)
|
|
{
|
|
if (myKeyToTablePointerMap.ContainsKey(row[myRightKey]))
|
|
{
|
|
((ParallelJoinTestTablePointer)myKeyToTablePointerMap[row[myRightKey]]).RightTableId = (int)row["id"];
|
|
}
|
|
else
|
|
{
|
|
myKeyToTablePointerMap[row[myRightKey]] = new ParallelJoinTestTablePointer(-1,(int)row["id"],-1);
|
|
|
|
}
|
|
|
|
}
|
|
}
|
|
|
|
if (myLeftTable != null && myRightTable != null)
|
|
{
|
|
//construct the schema for the table based upon the schemas of the two tables that are going to be joined together
|
|
int numColumns = myLeftTable.schema.columnInfo.Length;
|
|
|
|
|
|
ColumnInfo[] newColumnInfoArray = new ColumnInfo[numColumns * 2 + 1];
|
|
for (i = 0; i < numColumns; i++)
|
|
{
|
|
newColumnInfoArray[i] = new ColumnInfo("L." + myLeftTable.schema.columnNames[i], i, myLeftTable.schema.columnInfo[i].colType);
|
|
newColumnInfoArray[numColumns + i] = new ColumnInfo("R." + myRightTable.schema.columnNames[i], i + numColumns, myRightTable.schema.columnInfo[i].colType);
|
|
}
|
|
newColumnInfoArray[numColumns + i] = new ColumnInfo("id", numColumns + i, (typeof(System.Int64)));
|
|
|
|
IBlueTableSchema newSchema = new BlueTableSchema("Joined Table", newColumnInfoArray);
|
|
|
|
myJoinedTable = new BlueTable(newSchema);
|
|
addedRows = new BlueTable(newSchema);
|
|
droppedRows = new BlueTable(newSchema);
|
|
|
|
|
|
//now that we've created the table, we should have some leftover stuff from the side tables
|
|
//that we were storing until now. Add that stuff to the table.
|
|
BlueTable rowsToAdd = isLeft ? myRightTable : myLeftTable;
|
|
|
|
foreach (BlueRow row in rowsToAdd)
|
|
{
|
|
Object[] newRowObjects = new Object[newSchema.columnInfo.Length];
|
|
|
|
if (isLeft)
|
|
{
|
|
for (i = 0; i < numColumns; i++)
|
|
{
|
|
newRowObjects[i] = row[i];
|
|
newRowObjects[numColumns + i] = null;
|
|
|
|
}
|
|
newRowObjects[numColumns + i] = myIdToUse;
|
|
//store the fact that the row with this key value went to this id
|
|
|
|
((ParallelJoinTestTablePointer)myKeyToTablePointerMap[row[myLeftKey]]).JoinTableId = myIdToUse;
|
|
myIdToUse++;
|
|
|
|
}
|
|
else
|
|
{
|
|
for (i = 0; i < numColumns; i++)
|
|
{
|
|
newRowObjects[i] = null;
|
|
newRowObjects[numColumns + i] = row[i];
|
|
|
|
}
|
|
newRowObjects[numColumns + i] = myIdToUse;
|
|
//store the fact that the row with this key value went to this id
|
|
((ParallelJoinTestTablePointer)myKeyToTablePointerMap[row[myRightKey]]).JoinTableId = myIdToUse;
|
|
myIdToUse++;
|
|
}
|
|
|
|
BlueRow newRow = new BlueRow(newSchema, newRowObjects);
|
|
addedRows.addRow(newRow);
|
|
myJoinedTable.addRow(newRow);
|
|
|
|
}
|
|
|
|
|
|
}
|
|
}
|
|
else
|
|
{
|
|
addedRows = new BlueTable(myJoinedTable.schema);
|
|
droppedRows = new BlueTable(myJoinedTable.schema);
|
|
}
|
|
|
|
|
|
|
|
|
|
//setup is complete. At this point, we have to delete all the rows that were deleted, then add in the rows that were added in.
|
|
|
|
foreach (BlueRow row in dp.addedRows)
|
|
{
|
|
//check to see if this row has a match in the table already.
|
|
//If it does, get rid of that row in the table because it's OUT OF DATE
|
|
string joinColumn = isLeft ? myLeftKey : myRightKey;
|
|
|
|
if (myKeyToTablePointerMap.ContainsKey(row[joinColumn]))
|
|
{
|
|
//the value is already in the hashtable.
|
|
ParallelJoinTestTablePointer tp = (ParallelJoinTestTablePointer) myKeyToTablePointerMap[row[joinColumn]];
|
|
int rowOfExistingGuy = tp.JoinTableId;
|
|
|
|
if (rowOfExistingGuy != -1)
|
|
{
|
|
|
|
//remove this guy from the table
|
|
IBlueRow rowToDelete = myJoinedTable[rowOfExistingGuy];
|
|
myJoinedTable.deleteRow(rowToDelete);
|
|
droppedRows.addRow(rowToDelete);
|
|
tp.JoinTableId = -1;
|
|
}
|
|
|
|
|
|
|
|
}
|
|
}
|
|
|
|
|
|
//finally, fire off a packet, but only if you have enough information about the table to do so.
|
|
|
|
if (myJoinedTable != null)
|
|
{
|
|
DataPacket outPacket = new DataPacket(myJoinedTable,addedRows, new BlueTable(myJoinedTable.schema));
|
|
myFireEvent(outPacket);
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
}
|
|
|
|
|
|
}
|