Category: StreamInsight

//pull Austin URL from configuration file
string destination = ConfigurationManager.AppSettings["EventDestinationId"];
//build JSON message consisting of required headers, and data payload
string jsonPayload = "{\"DestinationID\":\"http:\\/\\/sample\\/\",\"Payload\":[{\"Key\":\"CustomerName\",\"Value\":\""+ txtRelatedParty.Text +"\"},{\"Key\":\"InteractionType\",\"Value\":\"Expense\"}],\"SourceID\":\"http:\\/\\/dummy\\/\",\"Version\":{\"_Build\":-1,\"_Major\":1,\"_Minor\":0,\"_Revision\":-1}}";

//update URL with JSON flag
string requestUrl = ConfigurationManager.AppSettings["AustinEndpoint"] + "json?batching=false";
HttpWebRequest request = HttpWebRequest.Create(requestUrl) as HttpWebRequest;

//set HTTP headers
request.Method = "POST";
request.ContentType = "application/json";

using (Stream dataStream = request.GetRequestStream())
 {
     string postBody = jsonPayload;

     // Create POST data and convert it to a byte array.
     byte[] byteArray = Encoding.UTF8.GetBytes(postBody);
     dataStream.Write(byteArray, 0, byteArray.Length);
  }

HttpWebResponse response = null;

try
{
    response = (HttpWebResponse)request.GetResponse();
 }
 catch (Exception ex)
 { }

public class FraudAlertEvent
    {
        public string CustomerName { get; set; }
        public string ExpenseDate { get; set; }
        public string AlertMessage { get; set; }
    }

public class TwilioPointOutputAdapter : TypedPointOutputAdapter
    {
        //member variables
        string acctId = string.Empty;
        string acctToken = string.Empty;
        string url = string.Empty;
        string phoneNum = string.Empty;
        string phoneOrMsg = string.Empty;
        TwilioRestClient twilioProxy;

        public TwilioPointOutputAdapter(AdapterConfig config)
        {
            //set member variables using values from runtime config values
            this.acctId = config.AccountId;
            this.acctToken = config.AuthToken;
            this.phoneOrMsg = config.PhoneOrMessage;
            this.phoneNum = config.TargetPhoneNumber;
            this.url = config.HandlerUrl;
        }

        ///
<summary> /// When the adapter is resumed by the engine, start dequeuing events again /// </summary>
        public override void  Resume()
        {
            DequeueEvent();
        }

        ///
<summary> /// When the adapter is started up, begin dequeuing events /// </summary>
        public override void  Start()
        {
            DequeueEvent();
        }

        ///
<summary> /// Function that pulls events from the engine and calls the Twilio service /// </summary>
        void DequeueEvent()
        {
		var twilioProxy = new TwilioRestClient(this.acctId, this.acctToken);

            while (true)
            {
                try
                {
                    //if the SI engine has issued a command to stop the adapter
                    if (AdapterState.Stopping == AdapterState)
                    {
                        Stopped();

                        return;
                    }

                    //create an event
                    PointEvent currentEvent = default(PointEvent);

                    //dequeue the event from the engine
                    DequeueOperationResult result = Dequeue(out currentEvent);

                    //if there is nothing there, tell the engine we're ready for more
                    if (DequeueOperationResult.Empty == result)
                    {
                        Ready();
                        return;
                    }

                    //if we find an event to process ...
                    if (currentEvent.EventKind == EventKind.Insert)
                    {
                        //append event-specific values to the Twilio handler service URL
                        string urlparams = "?val=0&action=Please%20look%20at%20" + currentEvent.Payload.CustomerName + "%20expenses";

                        //create object that holds call criteria
                        CallOptions opts = new CallOptions();
                        opts.Method = "GET";
                        opts.To = phoneNum;
                        opts.From = "+14155992671";
                        opts.Url = this.url + urlparams;

                        //if a phone call ...
                        if (phoneOrMsg == "phone")
                        {
                            //make the call
                            var call = twilioProxy.InitiateOutboundCall(opts);
                        }
                        else
                        {
                            //send an SMS message
                            var msg = twilioProxy.SendSmsMessage(opts.From, opts.To, "Fraud has occurred with " + currentEvent.Payload.CustomerName);
                        }
                    }
                    //cleanup the event
                    ReleaseEvent(ref currentEvent);
                }
                catch (Exception ex)
                {
                    throw ex;
                }
            }
        }
    }

private void CreateQueries()
{
		...

		//put inbound events into 30-second windows
     var custQuery = from i in allStream
          group i by new { Name = i.CustomerName, iType = i.InteractionType } into CustomerGroups
          from win in CustomerGroups.TumblingWindow(TimeSpan.FromSeconds(30), HoppingWindowOutputPolicy.ClipToWindowEnd)
          select new { ct = win.Count(), Cust = CustomerGroups.Key.Name, Type = CustomerGroups.Key.iType };

     //if there are more than two expenses for the same company in the window, raise event
     var thresholdQuery = from c in custQuery
                   where c.ct > 2 && c.Type == "Expense"
                   select new FraudAlertEvent
                   {
                          CustomerName = c.Cust,
                          AlertMessage = "Too many expenses!",
                          ExpenseDate = DateTime.Now.ToString()
                    };

      //call DeployQuery which instantiates StreamInsight Query
      Query query5 = DeployQuery(thresholdQuery, "Threshold Query");
       query5.Start();
		...
}

private Query DeployQuery(CepStream queryStream, string queryName)
{
      //setup Twilio adapter configuration settings
      var outputConfig = new AdapterConfig
       {
            AccountId = ConfigurationManager.AppSettings["TwilioAcctID"],
            AuthToken = ConfigurationManager.AppSettings["TwilioAcctToken"],
            TargetPhoneNumber = "+1111-111-1111",
            PhoneOrMessage = "phone",
            HandlerUrl = "http://twiliohandlerservice.ironfoundry.me/Handler.svc/Alert/Expense%20Fraud"
       };

      //add logging message
      lbMessages.Items.Add(string.Format("Creating new query '{0}'...", queryName));

      //define StreamInsight query that uses this output adapter and configuration
      Query query = queryStream.ToQuery(
            queryName,
            "",
            typeof(TwilioAdapterOutputFactory),
            outputConfig,
            EventShape.Point,
            StreamEventOrder.FullyOrdered);

      //return query to caller
      return query;
}

[ServiceContract]
    public interface IHandler
    {
        [OperationContract]
        [WebGet(
            BodyStyle = WebMessageBodyStyle.Bare,
            RequestFormat = WebMessageFormat.Xml,
            ResponseFormat = WebMessageFormat.Xml,
            UriTemplate = "Alert/{thresholdType}?val={thresholdValue}&action={action}"
            )]
        XmlElement GenerateHandler(string thresholdType, string thresholdValue, string action);
    }

        public XmlElement GenerateHandler(string thresholdType, string thresholdValue, string action)
        {
            string xml = string.Empty;

            if (thresholdValue == "0")
            {
                xml = "<!--?xml version='1.0' encoding='utf-8' ?-->" +
            "" +
            "" +
                "The " + thresholdType + " alert was triggered. " + action + "." +
                "" +
            "";
            }
            else
            {
                xml = "<!--?xml version='1.0' encoding='utf-8' ?-->" +
            "" +
            "" +
                "The " + thresholdType + " value is " + thresholdValue + " and has exceeded the threshold limit. " + action + "." +
                "" +
            "";
            }

            XmlDocument d = new XmlDocument();
            d.LoadXml(xml);

            return d.DocumentElement;
        }
    }