Storm系列（二）：使用Csharp创建你的第一个Storm拓扑（wordcount）

/// <summary>
/// Ack() will be called only when ack mechanism is enabled in spec file.
/// If ack is not supported in non-transactional topology, the Ack() can be left as empty function.
/// </summary>
/// <param name="seqId">Sequence Id of the tuple which is acked.</param>
public void Ack(long seqId)
{
Context.Logger.Info("Ack, seqId: {0}", seqId);
bool result = cachedTuples.Remove(seqId);
if (!result)
{
Context.Logger.Warn("Ack(), remove cached tuple for seqId {0} fail!", seqId);
}
}/// <summary>
/// Fail() will be called only when ack mechanism is enabled in spec file.
/// If ack is not supported in non-transactional topology, the Fail() can be left as empty function.
/// </summary>
/// <param name="seqId">Sequence Id of the tuple which is failed.</param>
public void Fail(long seqId)
{
Context.Logger.Info("Fail, seqId: {0}", seqId);
if (cachedTuples.ContainsKey(seqId))
{
string sentence = cachedTuples[seqId];
Context.Logger.Info("Re-Emit: {0}, seqId: {1}", sentence, seqId);
this.ctx.Emit("default", new List<object>() { sentence }, seqId);
}
else
{
Context.Logger.Warn("Fail(), can't find cached tuple for seqId {0}!", seqId);
}
}

private Context ctx;
private int msgTimeoutSecs;public Splitter(Context ctx)
{
Context.Logger.Info("Splitter constructor called");
this.ctx = ctx; // Declare Input and Output schemas
Dictionary<string, List<Type>> inputSchema = new Dictionary<string, List<Type>>();
inputSchema.Add("default", new List<Type>() { typeof(string) });
Dictionary<string, List<Type>> outputSchema = new Dictionary<string, List<Type>>();
outputSchema.Add("default", new List<Type>() { typeof(string), typeof(char) });
this.ctx.DeclareComponentSchema(new ComponentStreamSchema(inputSchema, outputSchema)); // Demo how to get stormConf info
if (Context.Config.StormConf.ContainsKey("topology.message.timeout.secs"))
{
msgTimeoutSecs = Convert.ToInt32(Context.Config.StormConf["topology.message.timeout.secs"]);
}
Context.Logger.Info("msgTimeoutSecs: {0}", msgTimeoutSecs);
}/// <summary>
/// Implements of delegate "newPlugin", which is used to create a instance of this spout/bolt
/// </summary>
/// <param name="ctx">Context instance</param>
/// <returns></returns>
public static Splitter Get(Context ctx)
{
return new Splitter(ctx);
}

void Prepare(Config stormConf, TopologyContext context);
void Execute(StormTuple tuple);

/// <summary>
/// The Execute() function will be called, when a new tuple is available.
/// </summary>
/// <param name="tuple"></param>
public void Execute(StormTuple tuple)
{
Context.Logger.Info("Execute enter"); string sentence = tuple.GetString(); foreach (string word in sentence.Split(' '))
{
Context.Logger.Info("Splitter Emit: {0}", word);
this.ctx.Emit("default", new List<StormTuple> { tuple }, new List<object> { word, word[] });
} Context.Logger.Info("Splitter Execute exit");
}public void Prepare(Config stormConf, TopologyContext context)
{
return;
}

using Storm;
using System;
using System.Collections.Generic;namespace StormSample
{
/// <summary>
/// The bolt "counter" uses a dictionary to record the occurrence number of each word.
/// </summary>
public class Counter : IBasicBolt
{
private Context ctx; private Dictionary<string, int> counts = new Dictionary<string, int>(); public Counter(Context ctx)
{
Context.Logger.Info("Counter constructor called"); this.ctx = ctx; // Declare Input and Output schemas
Dictionary<string, List<Type>> inputSchema = new Dictionary<string, List<Type>>();
inputSchema.Add("default", new List<Type>() { typeof(string), typeof(char) }); Dictionary<string, List<Type>> outputSchema = new Dictionary<string, List<Type>>();
outputSchema.Add("default", new List<Type>() { typeof(string), typeof(int) });
this.ctx.DeclareComponentSchema(new ComponentStreamSchema(inputSchema, outputSchema));
} /// <summary>
/// The Execute() function will be called, when a new tuple is available.
/// </summary>
/// <param name="tuple"></param>
public void Execute(StormTuple tuple)
{
Context.Logger.Info("Execute enter"); string word = tuple.GetString();
int count = counts.ContainsKey(word) ? counts[word] : ;
count++;
counts[word] = count; Context.Logger.Info("Counter Emit: {0}, count: {1}", word, count);
this.ctx.Emit("default", new List<StormTuple> { tuple }, new List<object> { word, count }); Context.Logger.Info("Counter Execute exit");
} /// <summary>
/// Implements of delegate "newPlugin", which is used to create a instance of this spout/bolt
/// </summary>
/// <param name="ctx">Context instance</param>
/// <returns></returns>
public static Counter Get(Context ctx)
{
return new Counter(ctx);
} public void Prepare(Config stormConf, TopologyContext context)
{
return;
}
}
}

using Storm;
using System;
using System.Linq;namespace StormSample
{
class Program
{
static void Main(string[] args)
{
if (args.Count() > )
{
string compName = args[]; try
{
if ("generator".Equals(compName))
{
ApacheStorm.LaunchPlugin(new newPlugin(Generator.Get));
}
else if ("splitter".Equals(compName))
{
ApacheStorm.LaunchPlugin(new newPlugin(Splitter.Get));
}
else if ("counter".Equals(compName))
{
ApacheStorm.LaunchPlugin(new newPlugin(Counter.Get));
}
else
{
throw new Exception(string.Format("unexpected compName: {0}", compName));
}
}
catch (Exception ex)
{
Context.Logger.Error(ex.ToString());
}
}
else
{
Context.Logger.Error("Not support local model.");
}
}
}
}