Use defer to close streams within consumers.