React 组件不会响应式重新渲染（Redux Observable + rxjs + rx-http-request）答案

【问题标题】：React component does not re-render reactively (Redux Observable + rxjs + rx-http-requst)React 组件不会响应式重新渲染（Redux Observable + rxjs + rx-http-request）
【发布时间】：2017-12-03 19:33:23
【问题描述】：

我有一个 API 端点，它以 'application/stream+json' 类型的响应返回用户列表。这些项目由换行符分隔。示例数据可见here。

组件

class UserList extends Component {
    componentDidMount() {
        const { fetchUsers } = this.props;
        fetchUsers();
    }

    render() {
        const { isFetching = false, users = [] } = this.props;
        if (isFetching) {
            return <Loader message="Users are loading..." />;
        }
        if (!users || users.length === 0) {
            return 'No users found.';
        }
        const children = users
            .map(user => <UserListItem key={user.id} user={user} />);
        return (
            <div className="UserList">
                <Paper>
                    <List>
                        <Subheader>Users</Subheader>
                        {children}
                    </List>
                </Paper>
            </div>
        );
    }
}

UserList.propTypes = {
    users: PropTypes.arrayOf(PropTypes.any),
    isFetching: PropTypes.bool.isRequired,
    fetchUsers: PropTypes.func.isRequired,
};

UserList.defaultProps = {
    users: [],
};

function mapStateToProps(state) {
    const { users, isFetching } = state.users;
    return {
        users,
        isFetching,
    };
}

function mapDispatchToProps(dispatch) {
    return {
        fetchUsers: bindActionCreators(actions.fetchUsers, dispatch),
    };
}

export default connect(mapStateToProps, mapDispatchToProps)(UserList);

减速器

const initialState = {
    users: [],
    isFetching: false,
};

function fetchUsers(state) {
    return {
        ...state,
        isFetching: true,
    };
}

function fetchUsersItemReceived(state, action) {
    const { user } = action;
    return {
        ...state,
        users: [...state.users, user],
        isFetching: false,
    };
}

export default function (state = initialState, action) {
    switch (action.type) {
    case actionTypes.FETCH_USERS_REQUEST:
        return fetchUsers(state);
    case actionTypes.FETCH_USERS_ITEM_RECEIVED:
        return fetchUsersItemReceived(state, action);
    default:
        return state;
    }
}

Action（解析器是 Streaming JSON Parser found here）

export function fetchUsers() {
    return {
        type: actionTypes.FETCH_USERS_REQUEST,
    };
}

function fetchUsersItemReceived(user) {
    return {
        type: actionTypes.FETCH_USERS_ITEM_RECEIVED,
        user,
    };
}


function fetchUsersSuccess() {
    return {
        type: actionTypes.FETCH_USERS_SUCCESS,
    };
}

function fetchUsersFailure(error) {
    return {
        type: actionTypes.FETCH_USERS_FAILURE,
        error,
    };
}

function getJsonStream(url) {
    const emitter = new Subject();
    const req$ = RxHR
        .get(url)
        .flatMap(resp => resp.body)
        .subscribe(
            (data) => {
                parser.write(data);
                parser.onValue = (value) => {
                    if (!parser.key) {
                        emitter.next(value);
                    }
                };
            },
            err => emitter.error(err),
            () => emitter.complete(),
        );
    return emitter;
}

export const fetchUsersEpic = action$ =>
    action$.ofType(actionTypes.FETCH_USERS_REQUEST)
        .concatMap(() => getJsonStream(`${api.API_BASE_URL}/user`))
        .map(user => fetchUsersItemReceived(user));

configureStore.js

const logger = createLogger();

const epicMiddleware = createEpicMiddleware(rootEpic);
const createStoreWithMiddleware = applyMiddleware(epicMiddleware, logger)(createStore);

export default function configureStore(initialState) {
    return createStoreWithMiddleware(rootReducer, initialState);
}

虽然列表组件应在收到每个项目后刷新，但在收到整个列表后刷新。有人可以指出代码中的阻塞点吗？

【问题讨论】：

你能把一个工作的 jsbin 或类似的东西放在一起吗？使用模拟数据而不是真实端点很好。有很多活动部件（对于一个 SO 问题），如果没有运行代码就很难知道。它还缺少您的组件如何连接到商店
@jayphelps 添加了缺失的状态流。明天将尝试将jsbin放在一起。我在史诗中偶然发现了一个可能使用 switchMap 的嫌疑人，将其更改为 concatMap 仍然没有乐趣。
好的，请随时联系我。此示例中的 FWIW switchMap 与 concatMap 看起来与您正在谈论的问题无关，因为它仅影响在前一个仍在流式传输时进入的新 FETCH_USERS_REQUEST。
@jayphelps 还没有时间设置 jsbin，尽管我尝试从数组中流式传输间隔模拟数据，并且 UI 会在每个项目上正确重新呈现。所以把字符流转成json对象的解析器一定是阻塞的。

标签： reactjs rxjs redux-observable

【解决方案1】：

这不是针对您的特定问题的解决方案（除非偶然哈哈），但我认为自定义 Observable 比 Subject 更适合这种情况。您也可以挂钩到 Parser 的错误回调。

认为稍后使用 rxjs 搜索流式 JSON 的其他人可能会发现这很方便（未经测试）

function streamingJsonParse(data$) {
  return new Observable((observer) => {
    const parser = new Parser();

    parser.onError = (err) => observer.error(err);
    parser.onValue = (value) => {
      if (!parser.key) {
        observer.next(value);
      }
    };

    // return the subscription so it's correctly
    // unsubscribed for us
    return data$
      .subscribe({
        next: (data) => parser.write(data),
        error: (e) => observer.error(e),
        complete: () => observer.complete()
      });
  });
}

function getJsonStream(url) {
  return RxHR
    .get(url)
    .mergeMap(resp => streamingJsonParse(resp.body));
}

当你有机会整合 jsbin 时告诉我！

【讨论】：

【解决方案2】：

原来问题出在 jsonParse 库上。切换到oboe.js 修复。使用 ”！”节点选择器选择多个根 JSON 元素我能够将字符流转换为用户对象流。

动作

function getJsonStream(url) {
    const emitter = new Subject();
    const emitter = new Subject();
    oboe(url)
        .node('!', (item) => {
            emitter.next(item);
        })
        .fail((error) => {
            emitter.error(error);
        });
    return emitter;
}

export const fetchUsersEpic = action$ =>
    action$.ofType(actionTypes.FETCH_USERS_REQUEST)
        .switchMap(() => getJsonStream(`${api.API_BASE_URL}/user`))
        .map(user => fetchUsersItemReceived(user));

【讨论】：