Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
using System;
using System.Text;
using Microsoft.ServiceBus.Messaging;
using System.Net;
using System.IO;
namespace StreamingAnalyticsEventPublisher
{
class MeetupRSVPEventSender
{
static string eventHubName = "meetup-rsvp-eh";
static string eventHubConnectionString = "<<YOUR AEH CONNECTION STRING>>";
static void Main(string[] args)
{
Console.WriteLine("Press Ctrl-C to stop the sender process");
Console.WriteLine("Press Enter to start process, now");
Console.ReadLine();
FetchAndPublishRSVPEvents();
}
static void FetchAndPublishRSVPEvents()
{
var eventHubClient = EventHubClient.CreateFromConnectionString(eventHubConnectionString, eventHubName);
try
{
try
{
// Source: Meetup API
Uri rsvpUri = new Uri("http://stream.meetup.com/");
System.Net.HttpWebRequest hwReq = (HttpWebRequest)WebRequest.Create(rsvpUri + "2/rsvps");
hwReq.Method = "GET";
HttpWebResponse hwResp = (HttpWebResponse)hwReq.GetResponse();
// Retrieve response stream and wrap in StreamReader
System.IO.Stream respStream = hwResp.GetResponseStream();
StreamReader strmReader = new StreamReader(respStream);
// Loop through response stream reading each line and publishing to event hub
string eachRSVPEvent = strmReader.ReadLine();
while (eachRSVPEvent != null)
{
// Publish only the events that have venue in them
//if (eachRSVPEvent.StartsWith("{\"venue\":"))
//{
// Send to event hub
eventHubClient.Send(new EventData(Encoding.UTF8.GetBytes(eachRSVPEvent)));
eachRSVPEvent = strmReader.ReadLine();
// Just for testing purposes
Console.WriteLine(eachRSVPEvent);
//}
}
strmReader.Close();
}
catch (WebException ex)
{
StreamReader strmReader = new StreamReader(ex.Response.GetResponseStream());
// Loop through response stream reading each line and publishing to event hub
string eachRSVPEvent = strmReader.ReadLine();
while (eachRSVPEvent != null)
{
// Publish only the events that have venue in them
if (eachRSVPEvent.StartsWith("{\"venue\":"))
{
// Send to event hub
eventHubClient.Send(new EventData(Encoding.UTF8.GetBytes(eachRSVPEvent)));
eachRSVPEvent = strmReader.ReadLine();
// Just for testing purposes
Console.WriteLine(eachRSVPEvent);
}
}
strmReader.Close();
}
}
catch (Exception exception)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine("{0} > Exception: {1}", DateTime.Now, exception.Message);
Console.ResetColor();
}
}
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.