=str #15755 #15756 First stab at input/output factories

* New naming based on Source and Sink

Some parts of this convoluted/wrong. Take a look and we can discuss.
wouldn't it be better to simply have:
`materializer.materialize(input, output, ops)` that returns the MaterializedFlow
In the end, if you add a new type of Sink or Source you must have a FlowMaterializer impl that understands those types?