Basic Source is not restartable

[JoshuaC215]

Currently, basic.Source.Stop() is not restart-able or checked by Close(). So, everything works as long as the same goroutine is writing to Input() and calling Close(), but stuff gets weird if you have multiple goroutines there.

This isn't blocking usage but it could generally be more robust. e.g. the batchproducer start/stop sync handling.