Files
spark/src/DataStreamSuite/DataStream/Primitives/TextFileStreams/TextReaderStream.cs
T

96 lines
2.7 KiB
C#

using System;
using System.Collections.Generic;
using System.IO;
namespace BCG
{
public class TextReaderStream : ViewStream
{
#region constructors
public TextReaderStream(string file)
{
myFile = file;
}
#endregion
#region ViewStream
protected override void myOnFirstSubscriber()
{
StreamReader reader = new StreamReader(myFile);
#region get schema
IBlueTableSchema? schema;
string? schemaStr;
if ((schemaStr = reader.ReadLine()) == null)
{
reader.Close();
return;
}
schema = myGetSchema(schemaStr);
#endregion
#region read rows and fire update
BlueTable adds = new BlueTable(schema);
BlueTable dels = new BlueTable(schema);
string? rowStr;
while ((rowStr = reader.ReadLine()) != null)
{
adds.addRow(myGetRow(schema, rowStr));
}
myFireEvent(new DataPacket(adds, adds, dels));
#endregion
reader.Close();
}
#endregion
#region helpers
private IBlueTableSchema myGetSchema(string schemaStr)
{
List<ColumnInfo> columnInfos = new List<ColumnInfo>();
List<string> columnInfoStrs = new List<string>(schemaStr.Split(new string[] { TextWriterStream.DELIM }, StringSplitOptions.None));
for(int i = 0; i < columnInfoStrs.Count; ++i)
{
string columnInfoStr = columnInfoStrs[i];
List<string> columnInfoStrParts = new List<string>(columnInfoStr.Split(new char[] { ':' }, StringSplitOptions.None));
if (columnInfoStrParts.Count != 2)
{
throw (new Exception(String.Format("Unable to decode schema string '{0}'", schemaStr)));
}
columnInfos.Add(new ColumnInfo(columnInfoStrParts[0], i, Type.GetType(columnInfoStrParts[1])));
}
return (new BlueTableSchema(columnInfos.ToArray(), ""));
}
private IBlueRow myGetRow(IBlueTableSchema schema, string rowStr)
{
List<string> valueStrs = new List<string>(rowStr.Split(new string[] { TextWriterStream.DELIM }, StringSplitOptions.None));
if (schema.columnInfo.Length != valueStrs.Count)
{
throw (new Exception(String.Format("Unable to decode row string '{0}'", rowStr)));
}
List<object> values = new List<object>();
for(int i = 0; i < schema.columnInfo.Length; ++i)
{
values.Add(Convert.ChangeType(valueStrs[i], schema.columnInfo[i].colType));
}
return (new BlueRow(schema, values.ToArray()));
}
#endregion
#region state
private string myFile;
#endregion
}
}