mod config;
mod errors;
mod events;
mod outstanding_transaction;
mod publish_request_type;
mod result;
mod state;

#[cfg(test)]
mod tests;

pub use self::config::ClientSessionConfig;
pub use self::errors::ClientSessionError;
pub use self::events::ClientSessionEvent;
pub use self::publish_request_type::PublishRequestType;
pub use self::result::ClientSessionResult;
pub use self::state::ClientState;

use self::outstanding_transaction::{OutstandingTransaction, TransactionPurpose};
use crate::rml::chunk_io::{ChunkDeserializer, ChunkSerializer, Packet};
use crate::rml::messages::{RtmpMessage, UserControlEventType};
use crate::rml::sessions::StreamMetadata;
use crate::rml::time::RtmpTimestamp;
use bytes::Bytes;
use rml_amf0::Amf0Value;
use std::collections::HashMap;
use std::mem;
use std::time::SystemTime;

type ClientResult = Result<Vec<ClientSessionResult>, ClientSessionError>;

/// A session that represents the client side of a single RTMP connection.
///
/// The `ClientSession` encapsulates the details of how a client interacts with a server.  This
/// includes how to connect to an application on the server, requesting publishing or playback,
/// and reacting to events and responses the server may send.
///
/// While RTMP connections can support multiple media operations simultaneously the `ClientSession`
/// has been designed for the main use case of one operation per connection.
///
/// Due to the way the header compression properties of the RTMP chunking protocol works,
/// is is required that:
///
/// * All bytes **after** the handshake has been completed are passed into the `ClientSession` in
/// the order they were received
/// * All responses generated by the session are sent to the server **in order**
/// * No extraneous bytes are passed into the session, and only bytes generated by the session are
/// sent to the server.
///
/// Any violation of these points have a high probability of causing RTMP chunk parsing errors
/// by either the `ClientSession` or the peer.
pub struct ClientSession {
	start_time: SystemTime,
	serializer: ChunkSerializer,
	deserializer: ChunkDeserializer,
	config: ClientSessionConfig,
	next_transaction_id: u32,
	outstanding_transactions: HashMap<u32, OutstandingTransaction>,
	current_state: ClientState,
	connected_app_name: Option<String>,
	active_stream_id: Option<u32>,
	peer_window_ack_size: Option<u32>,
	bytes_received: u64,
	bytes_received_since_last_ack: u32,
	connect_request_properties: HashMap<String, Amf0Value>,
}

impl ClientSession {
	/// Creates a new client session with the specified configuration
	///
	/// As part of the initial creation it automatically creates initial outbound RTMP messages,
	/// such as setting the outbound chunk size
	pub fn new(config: ClientSessionConfig) -> Result<(ClientSession, Vec<ClientSessionResult>), ClientSessionError> {
		let session = ClientSession {
			start_time: SystemTime::now(),
			serializer: ChunkSerializer::new(),
			deserializer: ChunkDeserializer::new(),
			next_transaction_id: 1,
			outstanding_transactions: HashMap::new(),
			current_state: ClientState::Disconnected,
			active_stream_id: None,
			connected_app_name: None,
			peer_window_ack_size: None,
			bytes_received: 0,
			bytes_received_since_last_ack: 0,
			connect_request_properties: HashMap::new(),
			config,
		};

		let results = Vec::new();

		Ok((session, results))
	}

	/// Takes in any number of bytes from the peer and processes them.  Any resulting responses or
	/// events are returned.
	pub fn handle_input(&mut self, bytes: &[u8]) -> ClientResult {
		let mut results = Vec::new();
		self.bytes_received += bytes.len() as u64;

		if let Some(peer_ack_size) = self.peer_window_ack_size {
			self.bytes_received_since_last_ack += bytes.len() as u32;
			if self.bytes_received_since_last_ack >= peer_ack_size {
				let ack_message = RtmpMessage::Acknowledgement {
					sequence_number: self.bytes_received_since_last_ack,
				};
				let ack_payload = ack_message.into_message_payload(self.get_epoch(), 0)?;
				let ack_packet = self.serializer.serialize(&ack_payload, false, false)?;

				self.bytes_received_since_last_ack = 0;
				results.push(ClientSessionResult::OutboundResponse(ack_packet));
			}
		}

		let mut bytes_to_process = bytes;
		loop {
			match self.deserializer.get_next_message(bytes_to_process)? {
				None => break, // no more messages
				Some(payload) => {
					let message = payload.to_rtmp_message()?;
					let mut message_results = match message {
						RtmpMessage::Acknowledgement { sequence_number } => {
							self.handle_acknowledgement(sequence_number)?
						}

						RtmpMessage::Amf0Command {
							command_name,
							transaction_id,
							command_object,
							additional_arguments,
						} => self.handle_amf0_command(
							command_name,
							transaction_id,
							command_object,
							additional_arguments,
						)?,

						RtmpMessage::Amf0Data { values } => self.handle_amf0_data(values, payload.message_stream_id)?,

						RtmpMessage::AudioData { data } => {
							self.handle_audio_data(payload.message_stream_id, data, payload.timestamp)?
						}

						RtmpMessage::VideoData { data } => {
							self.handle_video_data(payload.message_stream_id, data, payload.timestamp)?
						}

						RtmpMessage::UserControl {
							event_type,
							timestamp,
							stream_id,
							buffer_length,
						} => self.handle_user_control(event_type, timestamp, stream_id, buffer_length)?,

						RtmpMessage::WindowAcknowledgement { size } => self.handle_window_ack_size(size)?,

						RtmpMessage::SetChunkSize { size } => self.handle_set_chunk_size(size)?,

						_ => vec![ClientSessionResult::UnhandleableMessageReceived(payload)],
					};

					results.append(&mut message_results);
					bytes_to_process = &[];
				}
			}
		}

		Ok(results)
	}

	/// Sets extra name-value pairs to include in the `connect` command object,
	/// alongside `app` / `flashVer` / `objectEncoding`. This lets a client
	/// advertise enhanced-RTMP capabilities (e.g. `capsEx`) to the server.
	///
	/// Must be called before `request_connection`. Keys here override the built-in
	/// properties if they collide.
	pub fn set_connect_request_properties(&mut self, properties: HashMap<String, Amf0Value>) {
		self.connect_request_properties = properties;
	}

	/// Forms an RTMP message requesting a connection to the specified application on the server.
	/// An event will be raised when the request is accepted or rejected.
	pub fn request_connection(&mut self, app_name: String) -> Result<ClientSessionResult, ClientSessionError> {
		match self.current_state {
			ClientState::Disconnected => (),
			_ => {
				return Err(ClientSessionError::CantConnectWhileAlreadyConnected);
			}
		}

		let transaction_id = self.get_next_transaction_id();
		let transaction = OutstandingTransaction::ConnectionRequested {
			app_name: app_name.clone(),
		};
		self.outstanding_transactions.insert(transaction_id, transaction);

		let mut properties = HashMap::new();
		properties.insert("app".to_string(), Amf0Value::Utf8String(app_name));
		properties.insert(
			"flashVer".to_string(),
			Amf0Value::Utf8String(self.config.flash_version.clone()),
		);
		properties.insert("objectEncoding".to_string(), Amf0Value::Number(0.0));

		// Some implementations require a tcUrl to be sent up with the connection request
		match &self.config.tc_url {
			Some(tc_url) => {
				properties.insert("tcUrl".to_string(), Amf0Value::Utf8String(tc_url.clone()));
			}
			None => (),
		};

		// Caller-supplied properties (e.g. capsEx) override the built-ins on collision.
		for (key, value) in &self.connect_request_properties {
			properties.insert(key.clone(), value.clone());
		}

		let message = RtmpMessage::Amf0Command {
			command_name: "connect".to_string(),
			command_object: Amf0Value::Object(properties),
			additional_arguments: vec![],
			transaction_id: transaction_id as f64,
		};

		let payload = message.into_message_payload(self.get_epoch(), 0)?;
		let packet = self.serializer.serialize(&payload, false, false)?;

		Ok(ClientSessionResult::OutboundResponse(packet))
	}

	/// Starts the process of requesting playback on the server for the specified stream key.  An
	/// event will be raised when the request is accepted or rejected.  Once accepted we will
	/// receive audio, video, and metadata information via `ClientSessionEvent`s.
	pub fn request_playback(&mut self, stream_key: String) -> Result<ClientSessionResult, ClientSessionError> {
		match self.current_state {
			ClientState::Connected => (),
			_ => {
				return Err(ClientSessionError::SessionInInvalidState {
					current_state: self.current_state.clone(),
				});
			}
		}

		let transaction_id = self.get_next_transaction_id();
		let transaction = OutstandingTransaction::CreateStream {
			purpose: TransactionPurpose::PlayRequest { stream_key },
		};

		self.outstanding_transactions.insert(transaction_id, transaction);

		let message = RtmpMessage::Amf0Command {
			command_name: "createStream".to_string(),
			transaction_id: transaction_id as f64,
			command_object: Amf0Value::Null,
			additional_arguments: Vec::new(),
		};

		let payload = message.into_message_payload(self.get_epoch(), 0)?;
		let packet = self.serializer.serialize(&payload, false, false)?;

		Ok(ClientSessionResult::OutboundResponse(packet))
	}

	/// Starts the process of requesting to publish to the server on the specified stream key.  An
	/// event will be raised when the request is accepted or rejected.
	pub fn request_publishing(
		&mut self,
		stream_key: String,
		publish_type: PublishRequestType,
	) -> Result<ClientSessionResult, ClientSessionError> {
		match self.current_state {
			ClientState::Connected => (),
			_ => {
				return Err(ClientSessionError::SessionInInvalidState {
					current_state: self.current_state.clone(),
				});
			}
		}

		let transaction_id = self.get_next_transaction_id();
		let transaction = OutstandingTransaction::CreateStream {
			purpose: TransactionPurpose::PublishRequest {
				stream_key,
				request_type: publish_type,
			},
		};

		self.outstanding_transactions.insert(transaction_id, transaction);

		let message = RtmpMessage::Amf0Command {
			command_name: "createStream".to_string(),
			transaction_id: transaction_id as f64,
			command_object: Amf0Value::Null,
			additional_arguments: Vec::new(),
		};

		let payload = message.into_message_payload(self.get_epoch(), 0)?;
		let packet = self.serializer.serialize(&payload, false, false)?;

		Ok(ClientSessionResult::OutboundResponse(packet))
	}

	/// If currently playing on a stream key, this is used to tell the server we no longer want to
	/// play video from the stream.
	pub fn stop_playback(&mut self) -> ClientResult {
		// Validate we are in a state to do this
		match self.current_state {
			ClientState::Playing { .. } => (),
			ClientState::PlayRequested { .. } => (),
			_ => return Ok(Vec::new()), // Nothing to stop since we aren't performing playback
		}

		self.current_state = ClientState::Connected;
		match mem::replace(&mut self.active_stream_id, None) {
			None => Ok(Vec::new()), // Should never happen since we should always have a valid stream id
			Some(stream_id) => {
				let message = RtmpMessage::Amf0Command {
					command_name: "deleteStream".to_string(),
					transaction_id: 0.0, // always 0 per spec
					command_object: Amf0Value::Null,
					additional_arguments: vec![Amf0Value::Number(stream_id as f64)],
				};

				let payload = message.into_message_payload(self.get_epoch(), stream_id)?;
				let packet = self.serializer.serialize(&payload, false, false)?;
				Ok(vec![ClientSessionResult::OutboundResponse(packet)])
			}
		}
	}

	/// If currently publishing on a stream key, this is used to tell the server we no longer want
	/// to publish to that stream.
	pub fn stop_publishing(&mut self) -> ClientResult {
		// Validate we are in a state to do this
		match self.current_state {
			ClientState::Publishing { .. } => (),
			ClientState::PublishRequested { .. } => (),
			_ => return Ok(Vec::new()), // Nothing to stop since we aren't performing playback
		}

		self.current_state = ClientState::Connected;
		match mem::replace(&mut self.active_stream_id, None) {
			None => Ok(Vec::new()), // Should never happen since we should always have a valid stream id
			Some(stream_id) => {
				let message = RtmpMessage::Amf0Command {
					command_name: "deleteStream".to_string(),
					transaction_id: 0.0, // always 0 per spec
					command_object: Amf0Value::Null,
					additional_arguments: vec![Amf0Value::Number(stream_id as f64)],
				};

				let payload = message.into_message_payload(self.get_epoch(), stream_id)?;
				let packet = self.serializer.serialize(&payload, false, false)?;
				Ok(vec![ClientSessionResult::OutboundResponse(packet)])
			}
		}
	}

	/// Sends a ping request to the server.  An event will be raised when we get a response back
	pub fn send_ping_request(&mut self) -> Result<(Packet, RtmpTimestamp), ClientSessionError> {
		let current_epoch = self.get_epoch();
		let message = RtmpMessage::UserControl {
			event_type: UserControlEventType::PingRequest,
			buffer_length: None,
			stream_id: None,
			timestamp: Some(current_epoch.clone()),
		};

		let payload = message.into_message_payload(self.get_epoch(), 0)?;
		let packet = self.serializer.serialize(&payload, false, false)?;
		Ok((packet, current_epoch))
	}

	/// If publishing, this allows us to send encoder metadata to the server to send to all
	/// players.
	pub fn publish_metadata(&mut self, metadata: &StreamMetadata) -> Result<ClientSessionResult, ClientSessionError> {
		match self.current_state {
			ClientState::Publishing => (),
			_ => {
				return Err(ClientSessionError::SessionInInvalidState {
					current_state: self.current_state.clone(),
				});
			}
		}

		let active_stream_id = match self.active_stream_id {
			Some(x) => x,
			None => {
				return Err(ClientSessionError::NoKnownActiveStreamIdWhenRequired);
			}
		};

		let mut properties = HashMap::new();
		if let Some(x) = metadata.video_width {
			properties.insert("width".to_string(), Amf0Value::Number(x as f64));
		}

		if let Some(x) = metadata.video_height {
			properties.insert("height".to_string(), Amf0Value::Number(x as f64));
		}

		if let Some(x) = metadata.video_codec_id {
			properties.insert("videocodecid".to_string(), Amf0Value::Number(x as f64));
		}

		if let Some(x) = metadata.video_frame_rate {
			properties.insert("framerate".to_string(), Amf0Value::Number(x as f64));
		}

		if let Some(x) = metadata.video_bitrate_kbps {
			properties.insert("videodatarate".to_string(), Amf0Value::Number(x as f64));
		}

		if let Some(x) = metadata.audio_codec_id {
			properties.insert("audiocodecid".to_string(), Amf0Value::Number(x as f64));
		}

		if let Some(x) = metadata.audio_bitrate_kbps {
			properties.insert("audiodatarate".to_string(), Amf0Value::Number(x as f64));
		}

		if let Some(x) = metadata.audio_sample_rate {
			properties.insert("audiosamplerate".to_string(), Amf0Value::Number(x as f64));
		}

		if let Some(x) = metadata.audio_channels {
			properties.insert("audiochannels".to_string(), Amf0Value::Number(x as f64));
		}

		if let Some(x) = metadata.audio_is_stereo {
			properties.insert("stereo".to_string(), Amf0Value::Boolean(x));
		}

		if let Some(ref x) = metadata.encoder {
			properties.insert("encoder".to_string(), Amf0Value::Utf8String(x.clone()));
		}

		let message = RtmpMessage::Amf0Data {
			values: vec![
				Amf0Value::Utf8String("@setDataFrame".to_string()),
				Amf0Value::Utf8String("onMetaData".to_string()),
				Amf0Value::Object(properties),
			],
		};

		let payload = message.into_message_payload(self.get_epoch(), active_stream_id)?;
		let packet = self.serializer.serialize(&payload, false, false)?;

		Ok(ClientSessionResult::OutboundResponse(packet))
	}

	/// If publishing, this allows us to send video data to the server on the publishing stream.
	pub fn publish_video_data(
		&mut self,
		data: Bytes,
		timestamp: RtmpTimestamp,
		can_be_dropped: bool,
	) -> Result<ClientSessionResult, ClientSessionError> {
		match self.current_state {
			ClientState::Publishing => (),
			_ => {
				return Err(ClientSessionError::SessionInInvalidState {
					current_state: self.current_state.clone(),
				});
			}
		}

		let active_stream_id = match self.active_stream_id {
			Some(x) => x,
			None => {
				return Err(ClientSessionError::NoKnownActiveStreamIdWhenRequired);
			}
		};

		let message = RtmpMessage::VideoData { data };
		let payload = message.into_message_payload(timestamp, active_stream_id)?;
		let packet = self.serializer.serialize(&payload, false, can_be_dropped)?;
		Ok(ClientSessionResult::OutboundResponse(packet))
	}

	/// If publishing, this allows us to send audio data to the server on the publishing stream.
	pub fn publish_audio_data(
		&mut self,
		data: Bytes,
		timestamp: RtmpTimestamp,
		can_be_dropped: bool,
	) -> Result<ClientSessionResult, ClientSessionError> {
		match self.current_state {
			ClientState::Publishing => (),
			_ => {
				return Err(ClientSessionError::SessionInInvalidState {
					current_state: self.current_state.clone(),
				});
			}
		}

		let active_stream_id = match self.active_stream_id {
			Some(x) => x,
			None => {
				return Err(ClientSessionError::NoKnownActiveStreamIdWhenRequired);
			}
		};

		let message = RtmpMessage::AudioData { data };
		let payload = message.into_message_payload(timestamp, active_stream_id)?;
		let packet = self.serializer.serialize(&payload, false, can_be_dropped)?;
		Ok(ClientSessionResult::OutboundResponse(packet))
	}

	fn handle_video_data(&self, stream_id: u32, data: Bytes, timestamp: RtmpTimestamp) -> ClientResult {
		// PlayRequested state is allowed because some servers send video data prior to the
		// `NetStream.Play.Start` command.
		match self.current_state {
			ClientState::PlayRequested { .. } => (),
			ClientState::Playing { .. } => (),
			_ => {
				return Err(ClientSessionError::SessionInInvalidState {
					current_state: self.current_state.clone(),
				});
			}
		}

		// Validate we are active on the stream this message came from
		match self.active_stream_id {
			None => return Ok(Vec::new()), // not active on any stream
			Some(active_stream_id) if active_stream_id != stream_id => return Ok(Vec::new()), // not active on this stream
			Some(_) => (),
		}

		let event = ClientSessionEvent::VideoDataReceived { data, timestamp };
		Ok(vec![ClientSessionResult::RaisedEvent(event)])
	}

	fn handle_audio_data(&self, stream_id: u32, data: Bytes, timestamp: RtmpTimestamp) -> ClientResult {
		// PlayRequested state is allowed because some servers send audio data prior to the
		// `NetStream.Play.Start` command.
		match self.current_state {
			ClientState::PlayRequested { .. } => (),
			ClientState::Playing { .. } => (),
			_ => {
				return Err(ClientSessionError::SessionInInvalidState {
					current_state: self.current_state.clone(),
				});
			}
		}

		// Validate we are active on the stream this message came from
		match self.active_stream_id {
			None => return Ok(Vec::new()), // not active on any stream
			Some(active_stream_id) if active_stream_id != stream_id => return Ok(Vec::new()), // not active on this stream
			Some(_) => (),
		}

		let event = ClientSessionEvent::AudioDataReceived { data, timestamp };
		Ok(vec![ClientSessionResult::RaisedEvent(event)])
	}

	fn handle_amf0_data(&mut self, mut data: Vec<Amf0Value>, stream_id: u32) -> ClientResult {
		if data.len() == 0 {
			// No data so just do nothing
			return Ok(Vec::new());
		}

		// Validate we are active on the stream this message came from
		match self.active_stream_id {
			None => return Ok(Vec::new()), // not active on any stream
			Some(active_stream_id) if active_stream_id != stream_id => return Ok(Vec::new()), // not active on this stream
			Some(_) => (),
		}

		let first_element = data.remove(0);
		match first_element {
			Amf0Value::Utf8String(ref value) if value == "onMetaData" => self.handle_amf0_data_on_meta_data(data),

			_ => Ok(Vec::new()),
		}
	}

	fn handle_amf0_command(
		&mut self,
		name: String,
		transaction_id: f64,
		command_object: Amf0Value,
		additional_args: Vec<Amf0Value>,
	) -> ClientResult {
		match name.as_str() {
			"_result" => self.handle_amf0_command_success_result(transaction_id, command_object, additional_args),
			"_error" => self.handle_amf0_command_failed_result(transaction_id, command_object, additional_args),
			"onStatus" => self.handle_on_status_command(additional_args),

			_ => {
				let event = ClientSessionEvent::UnhandleableAmf0Command {
					command_name: name,
					additional_values: additional_args,
					command_object,
					transaction_id,
				};

				Ok(vec![ClientSessionResult::RaisedEvent(event)])
			}
		}
	}

	fn handle_amf0_command_failed_result(
		&mut self,
		transaction_id: f64,
		command_object: Amf0Value,
		mut additional_args: Vec<Amf0Value>,
	) -> ClientResult {
		let outstanding_transaction = match self.outstanding_transactions.remove(&(transaction_id as u32)) {
			Some(transaction) => transaction,
			None => {
				let event = ClientSessionEvent::UnknownTransactionResultReceived {
					additional_values: additional_args,
					command_object,
					transaction_id,
				};

				return Ok(vec![ClientSessionResult::RaisedEvent(event)]);
			}
		};

		match outstanding_transaction {
			OutstandingTransaction::ConnectionRequested { app_name: _ } => {
				let description = if additional_args.len() > 0 {
					if let Amf0Value::Object(mut properties) = additional_args.remove(0) {
						if let Some(Amf0Value::Utf8String(value)) = properties.remove("description") {
							value
						} else {
							"".to_string()
						}
					} else {
						"".to_string()
					}
				} else {
					"".to_string()
				};

				let event = ClientSessionEvent::ConnectionRequestRejected { description };
				Ok(vec![ClientSessionResult::RaisedEvent(event)])
			}

			OutstandingTransaction::CreateStream { purpose: _ } => {
				return Err(ClientSessionError::CreateStreamFailed);
			}
		}
	}

	fn handle_amf0_command_success_result(
		&mut self,
		transaction_id: f64,
		command_object: Amf0Value,
		additional_args: Vec<Amf0Value>,
	) -> ClientResult {
		let outstanding_transaction = match self.outstanding_transactions.remove(&(transaction_id as u32)) {
			Some(transaction) => transaction,
			None => {
				let event = ClientSessionEvent::UnknownTransactionResultReceived {
					additional_values: additional_args,
					command_object,
					transaction_id,
				};

				return Ok(vec![ClientSessionResult::RaisedEvent(event)]);
			}
		};

		match outstanding_transaction {
			OutstandingTransaction::ConnectionRequested { app_name } => {
				self.current_state = ClientState::Connected;
				self.connected_app_name = Some(app_name);

				let message = RtmpMessage::WindowAcknowledgement {
					size: self.config.window_ack_size,
				};
				let payload = message.into_message_payload(self.get_epoch(), 0)?;
				let packet = self.serializer.serialize(&payload, false, false)?;
				let event = ClientSessionEvent::ConnectionRequestAccepted;

				let chunk_size_packet = self
					.serializer
					.set_max_chunk_size(self.config.chunk_size, RtmpTimestamp::new(0))?;

				Ok(vec![
					ClientSessionResult::OutboundResponse(packet),
					ClientSessionResult::RaisedEvent(event),
					ClientSessionResult::OutboundResponse(chunk_size_packet),
				])
			}

			OutstandingTransaction::CreateStream { purpose } => {
				if additional_args.len() == 0 {
					return Err(ClientSessionError::CreateStreamResponseHadNoStreamNumber);
				}

				let stream_id = match additional_args[0] {
					Amf0Value::Number(number) => number as u32,
					_ => {
						return Err(ClientSessionError::CreateStreamResponseHadNoStreamNumber);
					}
				};

				self.active_stream_id = Some(stream_id);

				match purpose {
					TransactionPurpose::PlayRequest { stream_key } => {
						self.current_state = ClientState::PlayRequested;

						let buffer_message = RtmpMessage::UserControl {
							event_type: UserControlEventType::SetBufferLength,
							buffer_length: Some(self.config.playback_buffer_length_ms),
							stream_id: Some(stream_id),
							timestamp: None,
						};

						let buffer_payload = buffer_message.into_message_payload(self.get_epoch(), 0)?;
						let buffer_packet = self.serializer.serialize(&buffer_payload, false, false)?;

						let play_message = RtmpMessage::Amf0Command {
							command_name: "play".to_string(),
							transaction_id: 0.0,
							command_object: Amf0Value::Null,
							additional_arguments: vec![Amf0Value::Utf8String(stream_key)],
						};

						let play_payload = play_message.into_message_payload(self.get_epoch(), stream_id)?;
						let play_packet = self.serializer.serialize(&play_payload, false, false)?;

						Ok(vec![
							ClientSessionResult::OutboundResponse(buffer_packet),
							ClientSessionResult::OutboundResponse(play_packet),
						])
					}

					TransactionPurpose::PublishRequest {
						stream_key,
						request_type,
					} => {
						self.current_state = ClientState::PublishRequested;

						let publish_type_string = match request_type {
							PublishRequestType::Live => "live".to_string(),
							PublishRequestType::Record => "record".to_string(),
							PublishRequestType::Append => "append".to_string(),
						};

						let publish_message = RtmpMessage::Amf0Command {
							command_name: "publish".to_string(),
							transaction_id: 0.0,
							command_object: Amf0Value::Null,
							additional_arguments: vec![
								Amf0Value::Utf8String(stream_key),
								Amf0Value::Utf8String(publish_type_string),
							],
						};

						let publish_payload = publish_message.into_message_payload(self.get_epoch(), stream_id)?;
						let publish_packet = self.serializer.serialize(&publish_payload, false, false)?;
						Ok(vec![ClientSessionResult::OutboundResponse(publish_packet)])
					}
				}
			}
		}
	}

	fn handle_on_status_command(&mut self, mut arguments: Vec<Amf0Value>) -> ClientResult {
		if arguments.len() < 1 {
			return Err(ClientSessionError::InvalidOnStatusArguments);
		}

		let mut properties = match arguments.remove(0) {
			Amf0Value::Object(properties) => properties,
			_ => {
				return Err(ClientSessionError::InvalidOnStatusArguments);
			}
		};

		let code = match properties.remove("code") {
			Some(Amf0Value::Utf8String(code)) => code,

			_ => {
				return Err(ClientSessionError::InvalidOnStatusArguments);
			}
		};

		match code.as_ref() {
			"NetStream.Play.Start" => self.handle_play_start(),
			"NetStream.Publish.Start" => self.handle_publish_start(),

			x => {
				let event = ClientSessionEvent::UnhandleableOnStatusCode { code: x.to_string() };
				Ok(vec![ClientSessionResult::RaisedEvent(event)])
			}
		}
	}

	fn handle_play_start(&mut self) -> ClientResult {
		match self.current_state {
			ClientState::PlayRequested => (),
			_ => {
				return Err(ClientSessionError::SessionInInvalidState {
					current_state: self.current_state.clone(),
				});
			}
		};

		self.current_state = ClientState::Playing;

		let event = ClientSessionEvent::PlaybackRequestAccepted;
		Ok(vec![ClientSessionResult::RaisedEvent(event)])
	}

	fn handle_publish_start(&mut self) -> ClientResult {
		match self.current_state {
			ClientState::PublishRequested => (),
			_ => {
				return Err(ClientSessionError::SessionInInvalidState {
					current_state: self.current_state.clone(),
				});
			}
		};

		self.current_state = ClientState::Publishing;
		let event = ClientSessionEvent::PublishRequestAccepted;
		Ok(vec![ClientSessionResult::RaisedEvent(event)])
	}

	fn handle_amf0_data_on_meta_data(&mut self, mut data: Vec<Amf0Value>) -> ClientResult {
		if data.len() < 1 {
			// No data so ignore it
			return Ok(Vec::new());
		}

		let properties = match data.remove(0) {
			Amf0Value::Object(properties) => properties,
			_ => return Ok(Vec::new()), // malformed so ignore it
		};

		let mut metadata = StreamMetadata::new();
		metadata.apply_metadata_values(properties);

		let event = ClientSessionEvent::StreamMetadataReceived { metadata };
		Ok(vec![ClientSessionResult::RaisedEvent(event)])
	}

	fn handle_acknowledgement(&mut self, sequence_number: u32) -> ClientResult {
		let event = ClientSessionEvent::AcknowledgementReceived {
			bytes_received: sequence_number,
		};
		Ok(vec![ClientSessionResult::RaisedEvent(event)])
	}

	fn handle_window_ack_size(&mut self, size: u32) -> ClientResult {
		self.peer_window_ack_size = Some(size);
		Ok(Vec::new())
	}

	fn handle_user_control(
		&mut self,
		event_type: UserControlEventType,
		timestamp: Option<RtmpTimestamp>,
		_stream_id: Option<u32>,
		_buffer_length: Option<u32>,
	) -> ClientResult {
		match event_type {
			UserControlEventType::PingRequest => self.handle_ping_request(timestamp),
			UserControlEventType::PingResponse => self.handle_ping_response(timestamp),
			_ => Ok(Vec::new()),
		}
	}

	fn handle_ping_request(&mut self, timestamp: Option<RtmpTimestamp>) -> ClientResult {
		let message = RtmpMessage::UserControl {
			event_type: UserControlEventType::PingResponse,
			buffer_length: None,
			stream_id: None,
			timestamp,
		};

		let payload = message.into_message_payload(self.get_epoch(), 0)?;
		let packet = self.serializer.serialize(&payload, false, false)?;
		Ok(vec![ClientSessionResult::OutboundResponse(packet)])
	}

	fn handle_ping_response(&mut self, timestamp: Option<RtmpTimestamp>) -> ClientResult {
		let timestamp = timestamp.unwrap_or(RtmpTimestamp::new(0));
		let event = ClientSessionEvent::PingResponseReceived { timestamp };
		Ok(vec![ClientSessionResult::RaisedEvent(event)])
	}

	fn handle_set_chunk_size(&mut self, size: u32) -> ClientResult {
		self.deserializer.set_max_chunk_size(size as usize)?;
		Ok(Vec::new())
	}

	fn get_epoch(&self) -> RtmpTimestamp {
		match self.start_time.elapsed() {
			Ok(duration) => {
				let milliseconds = (duration.as_secs() * 1000) + (duration.subsec_nanos() as u64 / 1_000_000);

				// Casting to u32 should auto-wrap the value as expected.  If not a stream will probably
				// break after 49 days but testing shows it should wrap
				RtmpTimestamp::new(milliseconds as u32)
			}

			Err(_) => RtmpTimestamp::new(0), // Time went backwards, so just consider time as at epoch
		}
	}

	fn get_next_transaction_id(&mut self) -> u32 {
		let transaction_id = self.next_transaction_id;
		self.next_transaction_id += 1;
		transaction_id
	}
}
