class MapStageFactory implements ProcessingStageFactory<Stage.Map> {

    @Override
    public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Map stage) {
        Function<I, O> mapper = Casts.cast(stage.getMapper());
        Objects.requireNonNull(mapper);
        return source -> source.map(mapper::apply);
    }

}