Implementing twitter input in insight 2.1

Question

Hi,

I'm trying to understand how one would go around implementing twitter stream input in SI 2.1. Previously I've done this using the old approach defining a typedpointinput adapter and produce events myself as I would receive them in a loop that read the twitter stream api. Reading the news about SI 2.1 I'm thinking there is a simpler way to do this?

How would one go around implementing this in the new programming model available in SI 2.1. 

Thanks in advance.

Answer 1

Well, unless you have a compelling reason to update, then you probably should keep what is working. If it ain't broke, don't fix it, right?

However, that said, for a twitter feed, you'd want to do this as an Observable. As you get new items in from the feed, you would call OnNext on the Observer.

---

DevBiker (aka J Sawyer)
Microsoft MVP - Sql Server (StreamInsight)

---

---

If I answered your question, please mark as answer.
If my post was helpful, please mark as helpful.

Answer 2

Thanks for the quick reply DevBiker! 

Do you know of some sample code that implements this behavior (any source will do).

Answer 3

As an observable? The StreamInsight product team samples on CodePlex has some.

---

DevBiker (aka J Sawyer)
Microsoft MVP - Sql Server (StreamInsight)

---

---

If I answered your question, please mark as answer.
If my post was helpful, please mark as helpful.

Answer 4

I had a look but I'm not sure where. My current input adapter has a loop that simplied looks like this:

            JavaScriptSerializer ser = new JavaScriptSerializer();

            HttpWebRequest webRequest = (HttpWebRequest)WebRequest.Create(@"https://stream.twitter.com/1/statuses/filter.json?locations=-180,-90,180,90");
            webRequest.Credentials = new NetworkCredential("", "");
            webRequest.Timeout = -1;

            HttpWebResponse webResponse = (HttpWebResponse)webRequest.GetResponse();
            Encoding encode = System.Text.Encoding.GetEncoding("utf-8");

            using (StreamReader _responseStream = new StreamReader(webResponse.GetResponseStream(), encode))
            {
                while (true)
                {
                    string json = _responseStream.ReadLine();
                    Console.WriteLine(json);
                }
            }

This is where I produce my events in the old model, but I'm not sure how to convert it. I tried different things but I'm not sure how my source and sink would look if the payload was the json string. I hope you can help.

Answer 5

It would look something like the following:

void Main()
{
	var source = Application.DefineObservable<EventInfo<string>>(() => new TwitterObservable()); 
	var stream = source.ToPointStreamable(e => PointEvent.CreateInsert(e.StartTime, e.Payload));
}


public class TwitterObservable:IObservable<EventInfo<string>>, IDisposable{
	private IObserver<EventInfo<string>> _observer; 

	public IDisposable Subscribe(IObserver<EventInfo<string>> observer)
	{
		_observer = observer; 
		return this; 
	}

	private void ReadTwitter(){
	
			JavaScriptSerializer ser = new JavaScriptSerializer();

			HttpWebRequest webRequest = (HttpWebRequest)WebRequest.Create(@"https://stream.twitter.com/1/statuses/filter.json?locations=-180,-90,180,90");
			webRequest.Credentials = new NetworkCredential("", "");
			webRequest.Timeout = -1;

			HttpWebResponse webResponse = (HttpWebResponse)webRequest.GetResponse();
			Encoding encode = System.Text.Encoding.GetEncoding("utf-8");

			using (StreamReader _responseStream = new StreamReader(webResponse.GetResponseStream(), encode))
			{
				while (true)
				{
					string json = _responseStream.ReadLine();
					//Console.WriteLine(json);
					EventInfo<string> info = new EventInfo<string>(){
						StartTime = DateTimeOffset.Now, Payload = json};
					_observer.OnNext(info); 
				}
			}
	
	}
	
	public void Dispose(){
		//Cleanup.
	}

}
public class EventInfo<TPayload>
{
	public DateTimeOffset StartTime { get; set; }
	public TPayload Payload { get; set; }
}
// Define other methods and classes here

You'll want to do something a bit more sophisticated ... perhaps support multiple observers, use OnError and OnCompleted, etc. as well as your advance time settings and/or CTIs. But it should give you an idea of where to start.

---

DevBiker (aka J Sawyer)
Microsoft MVP - Sql Server (StreamInsight)

---

---

If I answered your question, please mark as answer.
If my post was helpful, please mark as helpful.

Answer 6

Thanks, that will get me started :-)

Answer 7

FYI, since the last Posts here there's also a new and great sample available parsing twitter streamingapi:

http://twitterbigdata.codeplex.com/

myself trying now to transform this sample to use oAuth (prereq. for consuming userstreams), wish i had more coding skills to get this to work ;)