Stream based Storm topology implementation.