Take message by ref in (filter)map_connect closures

[sbarral]

This avoids preemptive cloning when the closures don't consume the message, which is common when the filtering closure returns None.