refactored based on core.async