ibapi 5.4.0

//! Common StreamDecoder implementations for accounts module
//!
//! This module contains the StreamDecoder trait implementations that are shared
//! between sync and async versions, avoiding code duplication.

use crate::accounts::*;
use crate::messages::{IncomingMessages, Notice, ResponseMessage, REQUEST_VALIDATION_CODE};
use crate::subscriptions::{DecoderContext, StreamDecoder};
use crate::Error;

use super::{decoders, encoders};
use crate::common::error_helpers;

impl StreamDecoder<AccountSummaryResult> for AccountSummaryResult {
    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::AccountSummary, IncomingMessages::AccountSummaryEnd];

    fn decode(_context: &DecoderContext, message: &ResponseMessage) -> Result<Self, Error> {
        match message.message_type() {
            IncomingMessages::AccountSummary => Ok(AccountSummaryResult::Summary(decoders::decode_account_summary(message)?)),
            IncomingMessages::AccountSummaryEnd => Ok(AccountSummaryResult::End),
            _ => Err(Error::unexpected_response(message)),
        }
    }

    fn is_batch_end(&self) -> bool {
        matches!(self, AccountSummaryResult::End)
    }

    fn cancel_message(_server_version: i32, request_id: Option<i32>, _context: Option<&DecoderContext>) -> Result<Vec<u8>, Error> {
        let request_id = error_helpers::require_request_id(request_id)?;
        encoders::encode_cancel_account_summary(request_id)
    }
}

impl StreamDecoder<PnL> for PnL {
    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::PnL];

    fn decode(_context: &DecoderContext, message: &ResponseMessage) -> Result<Self, Error> {
        match message.message_type() {
            IncomingMessages::PnL => decoders::decode_pnl(message),
            _ => Err(Error::unexpected_response(message)),
        }
    }

    fn cancel_message(_server_version: i32, request_id: Option<i32>, _context: Option<&DecoderContext>) -> Result<Vec<u8>, Error> {
        let request_id = error_helpers::require_request_id(request_id)?;
        encoders::encode_cancel_pnl(request_id)
    }

    /// Live: after 321 TWS answers the cancel with 10185 (not subscribed).
    fn rejection_ends_request(notice: &Notice) -> bool {
        notice.code == REQUEST_VALIDATION_CODE
    }
}

impl StreamDecoder<PnLSingle> for PnLSingle {
    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::PnLSingle];

    fn decode(_context: &DecoderContext, message: &ResponseMessage) -> Result<Self, Error> {
        match message.message_type() {
            IncomingMessages::PnLSingle => decoders::decode_pnl_single(message),
            _ => Err(Error::unexpected_response(message)),
        }
    }

    fn cancel_message(_server_version: i32, request_id: Option<i32>, _context: Option<&DecoderContext>) -> Result<Vec<u8>, Error> {
        let request_id = error_helpers::require_request_id(request_id)?;
        encoders::encode_cancel_pnl_single(request_id)
    }

    /// Live: after 321 TWS answers the cancel with 10186 (not subscribed).
    fn rejection_ends_request(notice: &Notice) -> bool {
        notice.code == REQUEST_VALIDATION_CODE
    }
}

impl StreamDecoder<PositionUpdate> for PositionUpdate {
    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::Position, IncomingMessages::PositionEnd];

    fn decode(_context: &DecoderContext, message: &ResponseMessage) -> Result<Self, Error> {
        match message.message_type() {
            IncomingMessages::Position => Ok(PositionUpdate::Position(decoders::decode_position(message)?)),
            IncomingMessages::PositionEnd => Ok(PositionUpdate::PositionEnd),
            _ => Err(Error::unexpected_response(message)),
        }
    }

    fn is_batch_end(&self) -> bool {
        matches!(self, PositionUpdate::PositionEnd)
    }

    fn cancel_message(_server_version: i32, _request_id: Option<i32>, _context: Option<&DecoderContext>) -> Result<Vec<u8>, Error> {
        encoders::encode_cancel_positions()
    }
}

impl StreamDecoder<PositionUpdateMulti> for PositionUpdateMulti {
    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::PositionMulti, IncomingMessages::PositionMultiEnd];

    fn decode(_context: &DecoderContext, message: &ResponseMessage) -> Result<Self, Error> {
        match message.message_type() {
            IncomingMessages::PositionMulti => Ok(PositionUpdateMulti::Position(decoders::decode_position_multi(message)?)),
            IncomingMessages::PositionMultiEnd => Ok(PositionUpdateMulti::PositionEnd),
            _ => Err(Error::unexpected_response(message)),
        }
    }

    fn is_batch_end(&self) -> bool {
        matches!(self, PositionUpdateMulti::PositionEnd)
    }

    fn cancel_message(_server_version: i32, request_id: Option<i32>, _context: Option<&DecoderContext>) -> Result<Vec<u8>, Error> {
        let request_id = error_helpers::require_request_id(request_id)?;
        encoders::encode_cancel_positions_multi(request_id)
    }
}

impl StreamDecoder<AccountUpdate> for AccountUpdate {
    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[
        IncomingMessages::AccountValue,
        IncomingMessages::PortfolioValue,
        IncomingMessages::AccountUpdateTime,
        IncomingMessages::AccountDownloadEnd,
    ];

    fn decode(_context: &DecoderContext, message: &ResponseMessage) -> Result<Self, Error> {
        decoders::decode_account_update_message(message)
    }

    fn is_batch_end(&self) -> bool {
        matches!(self, AccountUpdate::End)
    }

    fn cancel_message(_server_version: i32, _request_id: Option<i32>, _context: Option<&DecoderContext>) -> Result<Vec<u8>, Error> {
        encoders::encode_cancel_account_updates()
    }
}

impl StreamDecoder<AccountUpdateMulti> for AccountUpdateMulti {
    const RESPONSE_MESSAGE_IDS: &'static [IncomingMessages] = &[IncomingMessages::AccountUpdateMulti, IncomingMessages::AccountUpdateMultiEnd];

    fn decode(_context: &DecoderContext, message: &ResponseMessage) -> Result<Self, Error> {
        match message.message_type() {
            IncomingMessages::AccountUpdateMulti => Ok(AccountUpdateMulti::AccountMultiValue(decoders::decode_account_multi_value(message)?)),
            IncomingMessages::AccountUpdateMultiEnd => Ok(AccountUpdateMulti::End),
            _ => Err(Error::unexpected_response(message)),
        }
    }

    fn is_batch_end(&self) -> bool {
        matches!(self, AccountUpdateMulti::End)
    }

    fn cancel_message(_server_version: i32, request_id: Option<i32>, _context: Option<&DecoderContext>) -> Result<Vec<u8>, Error> {
        let request_id = error_helpers::require_request_id_for(request_id, "encode cancel account updates multi")?;
        encoders::encode_cancel_account_updates_multi(request_id)
    }
}

#[cfg(test)]
#[path = "stream_decoders_tests.rs"]
mod tests;