[FLINK-1325] [streaming] Added clousure cleaning to streaming

Streaming update for @aljoscha's recent [PR](https://github.com/apache/incubator-flink/pull/269).

The functionality is there, the code could be a bit nicer but as @gyfora has already refactored the builder methods in the DataStream for the streaming [scala-api](https://github.com/mbalassi/incubator-flink/tree/scala-api) and not yet merged it this solution will be more straightforward to merge.

As this is part of the recently merged PR @aljoscha and myself are personally for adding it to the 0.8 release (with this codebase, not the merged one).