CREATE TABLE UserInformation(
    [UserId]         int NOT NULL,
    [FirstName]      nvarchar(32) NOT NULL,
    [LastName]       nvarchar(64) NOT NULL,
    [DisplayName]    nvarchar(128) NOT NULL,
    [Email]          nvarchar(256),
    [LastUpdated]    rowversion,
    CONSTRAINT [PK_UserInformation_UserId] PRIMARY KEY CLUSTERED ( [UserId] ASC )
)

INSERT INTO UserInformation (UserId, FirstName, LastName, DisplayName, Email) VALUES (1, 'Fred', 'Jones', 'Fred Jones', 'fred@email.com')
INSERT INTO UserInformation (UserId, FirstName, LastName, DisplayName, Email) VALUES (2, 'Bob', 'Foobar', 'Robert Foobar', 'bobrob@email.com')

/// <summary>
/// Configuration object for the SQL poller adapter
/// </summary>
public class SqlInputConfig
{
    /// <summary>
    /// Polling interval; how often to run the SQL command and check for new events
    /// </summary>
    public TimeSpan PollingInterval { get; set; } 

    /// <summary>
    /// Connection string to the data source
    /// </summary>
    public string ConnectionString { get; set; }

    /// <summary>
    /// Connection string to the data source
    /// </summary>
    public string ConnectionStringName { get; set; }

    /// <summary>
    /// SQL input mode - what type of command will be executed to retrieve data.  
    /// All commands must have a parameter corresponding to the TimestampField 
    /// option (rowversion data type)
        /// </summary>
    public SqlInputMode Mode { get; set; }

    /// <summary>
    /// The name of the field which stores the "last updated" time in the target
    /// table or stored procedure.  This fields needs to be of type rowversion
    /// </summary>
    public string TimestampField { get; set; }

    /// <summary>
    /// The name of the table or stored procedure
    /// </summary>
    public string Command { get; set; }

    public override string ToString()
    {
        StringBuilder sb = new StringBuilder();
        sb.AppendFormat("\tConnection String: {0}\r\n", ConnectionString);
        sb.AppendFormat("\tMode:              {0}\r\n", Mode);
        sb.AppendFormat("\tTimestamp Field:   {0}\r\n", TimestampField);
        sb.AppendFormat("\tSQL Command:       {0}\r\n", Command);
        return sb.ToString();
    }
}

/// <summary>
/// The valid SQL input mode types
/// </summary>
public enum SqlInputMode
{
    /// <summary>
    /// Invoke a stored procedure to retrieve values
    /// </summary>
    StoredProcedure,

    /// <summary>
    /// SELECT from a table to retrieve values
    /// </summary>
    Table,
}

/// <summary>
/// Implementation of an adapter factory for polling changing reference 
/// information from a SQL Server database
/// </summary>
public class SqlInputAdapterFactory : IInputAdapterFactory<SqlInputConfig>
{
    // All of our logging information will use this as the category name
    internal static readonly string APP_NAME = "Adapters.SqlInput";

    /// <summary>
    /// Called by the StreamInsight engine when creating an adapter.  Creates the
    /// appropriate adapter object.  Note that this adapter self-manages CTI events.
    /// </summary>
    /// <param name="configInfo">The SqlInputConfig class specified when the stream source is created</param>
    /// <param name="eventShape">The event shape; only Point events will be supported by this adapter</param>
    /// <param name="cepEventType">Description of the event type (fields, types, etc)</param>
    /// <returns>StreamInsight input adapter instance</returns>
    public InputAdapterBase Create(SqlInputConfig configInfo, EventShape eventShape,
        CepEventType cepEventType)
    {
            if (eventShape != EventShape.Point)
            throw new ArgumentException("Only Point events are supported for SQL Input polling reference adapter");

        InputAdapterBase ret = default(InputAdapterBase);

        switch(eventShape)
        {
            case EventShape.Point:
                ret = new SqlPointInputAdapter(TraceFactory.GetTracer(APP_NAME),
                    cepEventType, configInfo);
                break;
        }
        return ret;
    }

    /// <summary>
    /// This factory maintains no shared resources, nothing to dispose
    /// </summary>
    public void Dispose()
    { }
}

/// <summary>
/// Interface used to provide a callback between the SqlPoller and the
/// adapter instances
/// </summary>
internal interface IUntypedEventReceiver
{
    /// <summary>
    /// Given an untyped event payload, enqueue the contents as a CEP event
    /// </summary>
    /// <returns>Whether or not the event was successfully enqueued</returns>
    bool HandleEvent(UntypedEventPayload evt);

    /// <summary>
    /// Given a timestamp, advance stream time (enqueue a CTI)
    /// </summary>
    /// <param name="cti"></param>
    void FlushEvents(DateTimeOffset cti);
}

/// <summary>
/// Wrapper class to store key/value payloads.  Note: data needs to be
/// enqueued in the same order as the CepEventType ordinals in order to
/// be successfully enqueued
/// </summary>
internal class UntypedEventPayload
{
    /// <summary>
    /// Start time (receive time) of the event
    /// </summary>
    public DateTimeOffset StartTime { get; set; } 

    /// <summary>
    /// Array of key value pairs (the payload)
    /// </summary>
    public KeyValuePair<string, object>[] Data { get; set; }
}

/// <summary>
/// Implementation of an untyped input adapter for StreamInsight, designed to
/// retrieve reference (slow changing) information from a SQL server database
/// </summary>
internal sealed class SqlPointInputAdapter : PointInputAdapter,
    IUntypedEventReceiver
{
    #region "Private Variables"

    /// <summary>
    /// Tracing object; passed in from the factory
    /// </summary>
    private TraceLog trace;

    /// <summary>
    /// Configuration object; passed in from the factory
    /// </summary>
    private SqlInputConfig config;

    /// <summary>
    /// The CEP event type definition (this is the type that StreamInsight expects;
    /// the adapter will handle mapping between the SQL type and this type)
    /// </summary>
    private CepEventType eventType;

    /// <summary>
    /// The helper class responsible for managing interaction with SQL server
    /// </summary>
    private SqlPoller poller;

    #endregion

    /// <summary>
    /// Create a new SQL input adapter.  Note that this class is largely a wrapper around
    /// SqlPoller (where all of the heavy lifting happens).  Note: this class self-manages
    /// CTI events.
    /// </summary>
    /// <param name="trace">Initialized tracing object</param>
    /// <param name="eventType">CEP event type definition</param>
    /// <param name="config">Adapter configuration</param>
    public SqlPointInputAdapter(TraceLog trace,
        CepEventType eventType, SqlInputConfig config)
    {
        this.trace = trace;
        this.config = config;
        this.eventType = eventType;

        // Create the SQL poller object (this handles interaction with 
        // SQL server)
        this.poller = new SqlPoller(trace, eventType, config, this);
    }

    /// <summary>
    /// Start the adapter (initialize the poller)
    /// </summary>
    public override void Start()
    {
        this.poller.Start();
    }

    /// <summary>
    /// Resume the poller (if it has been paused)
    /// </summary>
    public override void Resume()
    {
        this.poller.Resume();
    }

    /// <summary>
    /// Dispose the poller
    /// </summary>
    public void Shutdown()
    {
        trace.LogInfo("Adapter is stopping - invoking shutdown");
        this.poller.Dispose();
        trace.LogInfo("Shutdown complete");
    }

    /// <summary>
    /// This is the callback for the SqlPoller to pass back new events
    /// data from the data source.  The adapter is responsible for 
    /// mapping the data into a CEP event and enqueueing
    /// </summary>
    /// <returns>Whether or not the event was successfully enqueued</returns>
    public bool HandleEvent(UntypedEventPayload evtPayload)
    {
        // Ensure that the adapter is in the running state
        if (AdapterState.Stopping == AdapterState)
        {
            Shutdown();
            Stopped();
        }

        // If the adapter is in a paused state ensure that we are not raising
        // more events
        if (AdapterState.Running != AdapterState)
        {
            this.poller.Pause();
            return false;
        }

        // Create a new untyped point event to hold the event payload
        PointEvent evt = CreateInsertEvent();
        if (evt == null)
        {
            trace.LogError("Could not allocate point event!");
            return false;
        }

        // Set the event timestamp to 'now'
        evt.StartTime = DateTimeOffset.UtcNow;

        // Fill in the event payload 
        foreach (var kv in evtPayload.Data)
        {
            int ordinal = this.eventType.Fields[kv.Key].Ordinal;
            evt.SetField(ordinal, kv.Value);
        }

        // If debug logging is enabled for this component, trace the contents
        if (trace.ShouldLogDebug())
        {
            trace.LogDebug("Raising SQL POINT input event - {0}",
                evtPayload.ToString());
        }

        // If the event cannot be enqueued, release the memory and signal that
        // the adapter is ready to process more events (via. Ready())
        if (EnqueueOperationResult.Full == Enqueue(ref evt))
        {
            ReleaseEvent(ref evt);
            Ready();
            return false;
        }

        return true;
    }

    /// <summary>
    /// Called by the SQL poller after a set of events has been received
    /// </summary>
    /// <param name="cti"></param>
    public void FlushEvents(DateTimeOffset cti)
    {
        // Ensure that the adapter is in the running state
        if (AdapterState.Stopping == AdapterState)
        {
            Shutdown();
            Stopped();
        }
        if (AdapterState.Running != AdapterState)
        {
            this.poller.Pause();
            return;
        }

        // Enqueue a CTI to advance time in the reference stream
        this.EnqueueCtiEvent(cti);
    }
}

/// <summary>
/// Called by the SQL poller after a set of events has been received
/// </summary>
/// <param name="cti"></param>
public void FlushEvents(DateTimeOffset cti)
{
    // Ensure that the adapter is in the running state
    if (AdapterState.Stopping == AdapterState)
    {
        Shutdown();
        Stopped();
    }
    if (AdapterState.Running != AdapterState)
    {
        this.poller.Pause();
        return;
    }

    // Enqueue a CTI to advance time in the reference stream
    this.EnqueueCtiEvent(cti);
}

/// <summary>
/// Handles periodically polling of a SQL data source looking for 
/// updated records
/// </summary>
internal sealed class SqlPoller : IDisposable
{
}

#region "Private Variables"

/// <summary>
/// Callback interface to the adapter instance to which we will 
/// publish new records
/// </summary>
private IUntypedEventReceiver recv;

/// <summary>
/// Adapter configuration
/// </summary>
private SqlInputConfig config;

/// <summary>
/// Tracing object
/// </summary>
private TraceLog trace;

/// <summary>
/// CEP event definition
/// </summary>
private CepEventType eventType;

/// <summary>
/// The SQL command used to retrieve new information
/// </summary>
private SqlCommand sqlCmd;

/// <summary>
/// Timer used to periodically poll the data source
/// </summary>
private System.Timers.Timer pollTimer;

/// <summary>
/// Dictionary storing the ordinal mapping between the SQL 
/// rows and the CEP event type
/// </summary>
private Dictionary<int, string> sqlToCepMap;

/// <summary>
/// The rowversion value expressed as an unsigned 64-bit integer
/// used to ask for "rowversions greater than this value"
/// </summary>
private System.UInt64 lastUpdated = 0;

/// <summary>
/// The ordinal of the rowversion column
/// </summary>
private int timestampOrdinal = -1;

/// <summary>
/// Synchronization object for the polling callback
/// </summary>
private object _lockObj = new object();
#endregion

/// <summary>
/// Create a new instance of the SQL poller, given a trace object,
/// CEP event type definition, adapter configuration and adapter
/// instance callback
/// </summary>
public SqlPoller(TraceLog trace, CepEventType eventType,
    SqlInputConfig config, IUntypedEventReceiver recv)
{
    this.eventType = eventType;
    this.trace = trace;
    this.config = config;
    this.recv = recv;

    trace.LogInfo("Creating SQL Server polling adapter component against \r\n\t{0}",
        config.ToString());

    // Create the requisite SQL command
    if (config.Mode == SqlInputMode.StoredProcedure)
        CreateSprocCmd();
    else if (config.Mode == SqlInputMode.Table)
        CreateTableCmd();

    // Create the type mapping
    InitializeTypeMapping();
}

/// <summary>
/// Create the SELECT command for direct table interaction
/// </summary>
private void CreateTableCmd()
{
    sqlCmd = new SqlCommand();
    sqlCmd.CommandText = String.Format("SELECT * FROM {0} WHERE {1} > @{1}",
        config.Command, config.TimestampField);
    sqlCmd.CommandType = System.Data.CommandType.Text;
    sqlCmd.Parameters.Add("@" + config.TimestampField, System.Data.SqlDbType.Timestamp);

}

/// <summary>
/// Create the SQL command for invoking a stored procedure
/// </summary>
private void CreateSprocCmd()
{
    sqlCmd = new SqlCommand();
    sqlCmd.CommandText = config.Command;
    sqlCmd.CommandType = System.Data.CommandType.StoredProcedure;
    sqlCmd.Parameters.Add("@" + config.TimestampField, System.Data.SqlDbType.Timestamp);
}

/// <summary>
/// Initialize the SQL type to CEP type mapping dictionary (ordinal
/// mapping) by invoking the lookup command in schema-only mode
/// </summary>
private void InitializeTypeMapping()
{
    trace.LogDebug("Initializing type mapping..");
    sqlToCepMap = new Dictionary<int, string>();

    using (var conn = new SqlConnection(config.ConnectionString))
    {
        sqlCmd.Connection = conn;
        sqlCmd.Connection.Open();

        // Retrieve only the command schema (no data)
        var rdr = sqlCmd.ExecuteReader(CommandBehavior.SchemaOnly);

        // Iterate through the list of fields, creating a mapping from
        // name -> ordinal
        var sqlOrdinals = new Dictionary<string, int>();
        for (int i = 0; i < rdr.FieldCount; i++)
        {
            var name = rdr.GetName(i);
            sqlOrdinals.Add(name, i);
            if (name == config.TimestampField)
                timestampOrdinal = i;
        }

        // Iterate through the CEP event type fields and create the mapping
        // [sql ordinal] -> [cep field name]
        foreach (var x in eventType.Fields)
        {
            if (!sqlOrdinals.ContainsKey(x.Key))
            {
                trace.LogWarn("Warning: did not find CEP field {0} (ord {1}) in SQL column list\r\n\tlist of SQL fields = {2}\r\n\tlist of CEP fields = {3}",
                    x.Key, x.Value.Ordinal, String.Join(", ", sqlOrdinals.Keys.ToArray()), String.Join(", ", eventType.Fields.Keys.ToArray()));
            }
            else
            {
                sqlToCepMap.Add(sqlOrdinals[x.Key], x.Key);
            }
        }
    }

    trace.LogDebug("Type mapping initialized with {0} fields", sqlToCepMap.Count);
}

#region "Lifecycle Management"

/// <summary>
/// Start the poller (activate the timer)
/// </summary>
public void Start()
{
    // Polling timer for retrieving SQL data
    pollTimer = new System.Timers.Timer();
    pollTimer.AutoReset = true;
    pollTimer.Elapsed += new System.Timers.ElapsedEventHandler(DoWork);
    pollTimer.Interval = (int)config.PollingInterval.TotalMilliseconds;
    pollTimer.Enabled = true;
}

public void Dispose()
{
    pollTimer.Dispose();
}

internal void Resume()
{
    pollTimer.Enabled = true;
}

internal void Pause()
{
    pollTimer.Enabled = false;
}

#endregion

/// <summary>
/// Polling timer callback method.  Polls the SQL data source to look for 
/// new / updated records
/// </summary>
private void DoWork(object sender, ElapsedEventArgs args)
{
    // Use a fail-fast mutex to avoid simultaneous invocations of the 
    // timer callback method.  This generally only happens when debugging
    // this function or a communication / timeout error.
    if (Monitor.TryEnter(_lockObj))
    {
        try
        {
            using (var conn = new SqlConnection(config.ConnectionString))
            {
                // Assign the new connection to the comamnd and open
                sqlCmd.Connection = conn;
                sqlCmd.Connection.Open();

                // Map the timestamp array back into a rowversion SQL parameter value
                var lastUpdatedBytes = BitConverter.GetBytes(lastUpdated);
                if (BitConverter.IsLittleEndian)
                    Array.Reverse(lastUpdatedBytes);
                sqlCmd.Parameters[0].Value = lastUpdatedBytes;

                // Execute the sql command to get the next set of data
                trace.LogDebug("Retrieving SQL data from {0} cmd {1} last Updated = {2}",
                    conn.DataSource, sqlCmd.CommandText, lastUpdated);
                var rdr = sqlCmd.ExecuteReader();
                System.UInt64 maxTimestamp = 0;

                while (rdr.Read())
                {
                    // Create an untyped event payload to wrap the SQL row
                    UntypedEventPayload evtPayload = new UntypedEventPayload();
                    List<KeyValuePair<string, object>> kvps =
                        new List<KeyValuePair<string, object>>();
                    foreach (var i in sqlToCepMap)
                    {
                        var name = i.Value;
                        var obj = rdr.GetValue(i.Key);
                        kvps.Add(new KeyValuePair<string, object>(name, obj));
                    }
                    evtPayload.Data = kvps.ToArray();
                    evtPayload.StartTime = DateTimeOffset.Now;

                    // Enqueue the event 
                    if (!recv.HandleEvent(evtPayload))
                    {
                        trace.LogDebug("Error: could not enqueue payload event");
                        return;
                    }

                    // Check for updated timestamp
                    if (timestampOrdinal > -1)
                    {
                        byte[] tsArray = (byte[])rdr.GetValue(timestampOrdinal);
                        if (BitConverter.IsLittleEndian)
                            Array.Reverse(tsArray);

                        // Convert the timestamp (8-byte array) into an unsigned 64-bit
                        // integer for easy comparison
                        System.UInt64 ts = BitConverter.ToUInt64(tsArray, 0);
                        if (ts > maxTimestamp)
                            maxTimestamp = ts;
                    }
                }

                // After all events have been enqueued, advance stream 
                // time by enqueueing a CTI
                recv.FlushEvents(DateTimeOffset.UtcNow);

                // Advance the last updated time (all events have been enqueued)
                lastUpdated = maxTimestamp;
            }
        }
        catch (Exception ex0)
        {
            trace.LogException(ex0, "Error in SQL retrieve");
        }
        finally
        {
            Monitor.Exit(_lockObj);
        }
    }
}    

/// <summary>
/// Sample activity event.  ActivityEventFactory can create 
/// random ActivityEvent objects
/// </summary>
[EventGenerator(typeof(ActivityEventFactory))]
public class ActivityEvent
{
    public int UserId { get; set; }
    public string Activity { get; set; }
    public DateTime Timestamp { get; set; }
    public string Status { get; set; }
}

/// <summary>
/// Factory for creating random activity events 
/// </summary>
public class ActivityEventFactory :
    IRandomEventFactory<ActivityEvent>
{
    private Random rand = new Random();

    /// <summary>
    /// The range of user IDs
    /// </summary>
    public int[] userIds = new int[]
    {
        1, 2, 3, 4
    };

    /// <summary>
    /// Range of activity names
    /// </summary>
    public string[] activityNames = new string[]
    {
        "Login", "Logout", "BuyStuff", "SellStuff"
    };

    public string[] statusNames = new string[]
    {
        "Success", "Fail", "FileNotFound"
    };

    ///
    public int GetRandomUser()
    {
        return userIds[rand.Next(userIds.Length - 1)];
    }

    public string GetRandomActivity()
    {
        return activityNames[rand.Next(activityNames.Length - 1)];
    }

    public string GetRandomStatus()
    {
        return statusNames[rand.Next(statusNames.Length - 1)];
    }

    public ActivityEvent GetRandomEvent()
    {
        return new ActivityEvent()
        {
            UserId = GetRandomUser(),
            Activity = GetRandomActivity(),
            Timestamp = DateTime.UtcNow,
            Status = GetRandomStatus()
        };
    }
}

<?xml version="1.0"?>
<configuration>
  <configSections>
    <section name="nlog" type="NLog.Config.ConfigSectionHandler, NLog"/>
  </configSections>

  <connectionStrings>
    <add name="TestDatabase" connectionString="Data Source=(local);Initial Catalog=RefDatabase;Integrated Security=True"/>
  </connectionStrings>

  <nlog>
    <!-- Include the debugging version of the logging configuration -->
    <include file="${basedir}/nlog.config"/>
  </nlog>

  <startup>
    <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.0"/>
  </startup>
</configuration>

// We want time to advance as new activity updates arrive.  In order to do this
// we need to link the reference stream's CTIs to the activity stream via a 
// time import setting
var timeImportSettings = new AdvanceTimeSettings(null,
    new AdvanceTimeImportSettings(activityStreamName),
    AdvanceTimePolicy.Adjust);

// Create a reference stream of database updates
var connString = ConfigurationManager.ConnectionStrings["TestDatabase"];
var referenceCfg = new SqlInputConfig()
{
    ConnectionString = connString.ConnectionString,
    Mode = SqlInputMode.Table,
    Command = "UserInformation",
    PollingInterval = TimeSpan.Parse("00:00:30"),
    TimestampField = "LastUpdated",
};

// Use the time import settings to link CTI values from the activity
// stream
var referenceStream = CepStream<UserEvent>.Create(
    "referenceStream", typeof(SqlInputAdapterFactory),
    referenceCfg, EventShape.Point, timeImportSettings);

// Convert the reference point stream into a reference signal
var referenceSignal = referenceStream
    .AlterEventDuration(e => TimeSpan.MaxValue)
    .ClipEventDuration(referenceStream, (e1, e2) => (e1.UserId == e2.UserId));
   

var outputCfg = new TracerConfig()
{
    DisplayCtiEvents = false,
    SingleLine = true,
    TracerKind = TracerKind.Console,
    TraceName = "OUTPUT"
};

var q = referenceStream.ToQuery(cepApp,
    "enrichedData", "",
    typeof(TracerFactory), outputCfg,
    EventShape.Point, StreamEventOrder.FullyOrdered);

// Join the two streams to enrich activity data with user details
var enrichedStream =
    from a in activityStream
    join r in referenceSignal on a.UserId equals r.UserId
    select new
    {
        UserId = a.UserId,
        Activity = a.Activity,
        Status = a.Status,
        Name = r.DisplayName
    };

<rules>
  <!-- Individual component log configuration -->
  <logger name="Adapters.OutputTracer" minlevel="Info"  writeTo="file,console" />
  <logger name="Adapters.SqlInput"     minlevel="Debug"  writeTo="file,console" />
  <logger name="Adapters.RandomData"     minlevel="Info"  writeTo="file,console" />
</rules>