[This is preliminary documentation and is subject to change.]

Custom vertices are defined by inheriting from the VertexTTime class, which is a class wrapping the common logic for all vertices that may be brought in to existence in a Naiad stage. The generic parameter TTime denotes the type of Naiad timestamp the vertex uses, which will often be a generic parameter for your custom vertex, too. We'll get to the specific features of the Vertex class later. An example vertex with one input and one output might look like the following

// Single input, single output Naiad vertex. Each vertex must specify a time type.
public class ExampleVertex<TRecord, TTime> : Vertex<TTime>
    where TTime : Time<TTime>
{
    // each received message is simply forwarded to any listeners.
    public void OnReceive(Message<TRecord, TTime> message)
    {
        foreach (var listener in this.Listeners)
            recipient.Send(message);
    }

    // list of output recipients, each of which require all sent records.
    public readonly List<SendWire<TRecord, TTime>> Listeners;

    // each vertex requires its index in the stage, and the stage itself.
    public ExampleVertex(int index, Stage<TTime> stage)
        : base(index, stage)
    {
        this.Listeners = new List<SendWire<TRecord, TTime>>();
    }
}

This vertex isn't especially interesting yet, and it isn't even clear how this list of recipients is populated. However, assuming that we sort that out, the vertex's behavior is specified: for each received message, it sends that message to every one in its list of recipients.

To complete this example, we have to assemble a stage consisting of many of these vertices, and describe how the inputs and outputs are connected. The standard pattern is to first create a new stage from a vertex factory, then add any number of inputs (defined by a message callback and partitioning requirement), and finally add any number of outputs (defined by a registration callback and partitioning guarantee). The example looks like:

// constructs a stage of our ExampleVertex, and returns the associated output stream.
public static Stream<TRecord, TTime> MakeStage(Stream<TRecord, TTime> stream,
                                                     Expression<Func<TRecord, int>> inputPartitionBy,
                                                     Expression<Func<TRecord, int>> outputPartitionBy,
                                                     string name)
{
    // first we define the innards of the stage, supplying a context, a vertex factory, and a friendly name.
    var stage = Foundry.NewStage(stream.Context, (index, parent) => new ExampleVertex<TRecord, TTime>(index, parent), name);

    // each new input requires a source of data, a (message, vertex) callback, and a partitioning requirement.
    var input = stage.NewInput(stream, (message, vertex) => vertex.OnReceive(message), inputPartitionBy);

    // each new output requires a (listener, vertex) callback, and a partitioning guarantee.
    var output = stage.NewOutput((listener, vertex) => vertex.Listeners.Add(listener), outputPartitionBy);

    return output;
}

In more detail, the NewStageTVertex, TTime(TimeContextTTime, FuncInt32, StageTTime, TVertex, String) method is what Naiad uses to assemble a new stage, taking a TimeContextTTime from which the stage understands its place in the dataflow graph, a factory capable of producing vertices for each vertex index and stage, and a tasteful name to use in describing the vertex. This stage will contain many vertices constructed by the factory, but does not yet have any incoming or outgoing edges. To add a new input, the NewInputTRecord(StreamTRecord, TTime, ActionMessageTRecord, TTime, TVertex, ExpressionFuncTRecord, Int32) method needs a source of data, a callback for each vertex supplied as an action on messages and vertices, and a partitioning requirement. Naiad will make sure that all records passed along the input will be partitioned according to the requirement, in that two records evaluating to the same value will arrive at the same vertex instance. Once all inputs are added we can add outputs using NewOutputR(FuncTVertex, VertexOutputR, TTime, ExpressionFuncR, Int32), which requires a registration callback (essentially what the vertex should do when another vertex expresses interest in its output), and a partitioning guarantee. Naiad uses the guarantee to pipeline communication when an interested consumer has a requirement matching the guarantee.