/// ****************************************************************************
/// @file   
/// @author Dr.Neeton
/// @date   2007/X/X - 2016/11/29
/// 
/// @brief  ソケットクラス 送信スレッド関連。
/// 
/// ****************************************************************************

#include "CPL_WIN_WS_SocketTCP.h"

namespace CPL {
namespace WIN {


// _____________________________________________________________________________
/// @brief コンストラクタ
/// 
/// @param[in] in_ptr_socket
///   : 送信するソケットのポインタを指定します。
/// @param[in] in_buffer_table
///   : 送信するバッファ(ByteBuffer)を指すスマートポインタの配列を指定します。
/// @param[in] in_table_num
///   : in_buffer_tableの配列数を指定します。
/// @param[out] out_ptr_ID
///   : IDを取得する場合、取得する変数(U32型)のポインタを指定を指定します。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
SocketTCP::ThreadSend::ThreadSend
( SocketTCP*         in_ptr_socket
, SpByteBuffer const in_buffer_table[]
, int                in_table_num
, U32*               out_ptr_ID
) : m_ptr_socket( in_ptr_socket )
  , m_flag_sending( false )
  , m_cache_buffer( CPL_WIN_WS_SOCKET_TEMP_BUF_SIZE ) // 領域のみ確保。
  , m_send_size( 0 )
  , m_complete_seg_num( -1 )
  , m_flag_life_thread( false )
{
  using FMH = Socket_i_::FragmentHeader;
  FMH tmp_header;

  m_cache_buffer.AddEnd( (U8_cptr)&tmp_header, sizeof(tmp_header) );

  auto id = AddData( in_buffer_table, in_table_num );
  if( out_ptr_ID )
  {
    *out_ptr_ID = id;
  }
}


// _____________________________________________________________________________
/// @brief 最終的（最も単純）な送信関数。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
int SocketTCP::ThreadSend::zzf_Send( SendParam& io_send_param ) const
{
  int send_size =
    io_send_param.in_length > INT_MAX ?
      INT_MAX : (int)io_send_param.in_length;

  int ret =
    send
      ( m_ptr_socket->GetSOCKET()
      , (char_cptr)io_send_param.in_buf
      , send_size
      , 0
      );

  CPL_BREAK( ret < 0 );

#ifdef CPL_DBGSW_WINSOCK_send
  const int length_max = 128;
  char tmp_send_str[ length_max * 2 + 1 ] = { 0 };
  int length = Min<int>( ret, length_max );
  Core::strfunc<char>::binary_to_x16
    ( tmp_send_str, sizeof(tmp_send_str)
    , io_send_param.in_buf
    , length
    );
  CPL_WIN_OUTPUT_EX_A( "send( size : %d ) : %s\n", (int)ret, &tmp_send_str[0] );
#endif

  return ret;
}


// _____________________________________________________________________________
/// @brief 基本送信処理
/// 
/// @param[in,out] io_send_param : 送信パラメタオブジェクトを指定します。
/// 
/// @retval 0 : ソケットが閉じられた。
/// @retval SOCKET_ERROR : エラー。
/// @return 送信バイト数を返す。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
S64 SocketTCP::ThreadSend::zzf_SendEx( SendParam& io_send_param ) const
{
  // 送信ループ
  S64  send_size = io_send_param.in_length;
  S64  send_count = 0;
  while( true )
  {
    int ret = zzf_Send( io_send_param ); // 書き込み。
    if( ret > 0 )
    {
      send_size -= ret;
      send_count += ret;
      if( io_send_param.out_size_sending )
      {
        *io_send_param.out_size_sending += ret;
      }
      if( send_size > 0 ) // まだ送信するデータが残っている場合
      {
        continue;
      }
      return send_count;
    }
    else // エラーの場合
    {
      int err = WIN::OutputWSALastErrorMessage();
      io_send_param.out_error = err;
      // フラグチェック
      if( ret == 0 ) // ソケットが閉じられた。
      {
        return 0;
      }
      //
      if( err == WSAEWOULDBLOCK ) // 送信バッファに空きが無い場合。
      {
        // ソケットにイベント（空き）が来るまで待機。
        EventParam ep;
        ep.in_event_flags = FD_WRITE | FD_CLOSE;
        int flag = m_ptr_socket->WaitEvents( ep ); // イベント待機。
        if( flag == SOCKET_ERROR ) // エラー
        {
          io_send_param.out_error = ep.out_error;
          return SOCKET_ERROR;
        }
        long event_flag = ep.out_event_flags;
        //
        if( event_flag <= 0 ) // タイムアウトor離脱イベントがシグナル化された場合。
        {
          return 0; // 終了。
        }
        if( event_flag == FD_WRITE ) // 書き込み可能になった。
        {
          continue;
        }
        else
        if( event_flag == FD_CLOSE ) // ソケットが閉じられた。
        {
          return 0;
        }
        else // それ以外ならばエラーとして抜ける。
        {
          io_send_param.out_error = ep.out_error;
          break;
        }
      }
    }

    break; // エラーとして抜ける
  }

  return SOCKET_ERROR; // エラー値を返す。
}


// _____________________________________________________________________________
/// @brief キャッシュの内容の送信
/// 
/// @retval 0 : ソケットが閉じられた。
/// @retval SOCKET_ERROR : エラー。
/// @return 送信バイト数を返す。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
S64 SocketTCP::ThreadSend::zzf_SendQueue() const
{
  DWORD t_base = timeGetTime();
  
  S64 size_count = 0;
  //size_t block_size = 0;
  RequestInfoBuffer req_info(0);
  
  while( !IsCancelThread() ) //
  {
    CPL_EXCTION_( m_mutex_data, m_mutex_request ) // リクエストバッファとデータバッファのロック
    {
      // リクエスト情報キューが空の場合、抜ける。
      if( m_send_request_queue.empty() )
      {
        // バッファキャッシュが残っていたら（本来ありえない）すべて削除する。
        while( m_send_data_queue.IsEmpty() == false )
        {
          m_send_data_queue.Pop();
        }
        break;
      }
      // バッファキャッシュが空の場合、抜ける。
      if( m_send_data_queue.IsEmpty() )
      {
        break;
      }
      // 送信中フラグONのタイミングはここで。
      m_flag_sending = true;

      // ブロックサイズ取得。
      //block_size = m_size_queue.front();
      req_info = m_send_request_queue.front();
    }
    // ヘッダ情報追記。
    SendParam sp;
    sp.in_buf    = (U8_cptr)req_info.get();
    sp.in_length = req_info.GetBufferSize();
    // 暗号化される可能性があるので、必要なパラメータを保存する。
    RequestInfoBuffer backup_req_info;
    req_info.CreateClone( backup_req_info );
    //
    cref<std::string> password = m_ptr_socket->GetSendPassword();
    if( !password.empty()  &&  password != "" )
    {
      // ヘッダは独立で暗号化する（基本情報のみ）
      Core::Encryption
        ( (char*)req_info.get()
        , sizeof(RequestInfo)
        , &password[0]
        , static_cast<int>(password.length())
        );
    }
    m_complete_seg_num = 0;
    //U32 ret = zzf_SendQueue_Push( sp );//zzf_SendEx( sp );
    S64 ret = zzf_SendEx( sp ); // ヘッダはキャッシュにプッシュせず直接送信。
    if( ret <= 0 )
    {
      //io_send_param.out_error = sp.out_error;
      m_flag_sending = false;
      m_complete_seg_num = -1;
      return ret;
    }
    size_count += ret;

    // データ本体の送信
    SpByteBuffer sp_temp_buf;
    S64 end_size = backup_req_info->m_size;
    m_send_size = 0;

    while( size_count < end_size )
    {
      CPL_EXCTION_( m_mutex_data ) // キューの取得から削除まで一連して行うため、ロックする。
      {
        // キャッシュからバッファを取得。
        sp_temp_buf = m_send_data_queue.Pop();
      }

      // データ本体
      sp.in_buf           = &(*sp_temp_buf)[0];
      sp.in_length        = static_cast<int>(backup_req_info->m_segment[ m_complete_seg_num ].m_size);
      sp.out_size_sending = &m_send_size;
      S64 push_size = zzf_SendQueue_Push( sp );
      if( push_size == 0 )
      {
        m_flag_sending = false;
        m_complete_seg_num = -1;
        return 0;
      }
      size_count += push_size;
      ++m_complete_seg_num;
    }

    m_flag_sending = false;
    m_send_request_queue.pop_front();

#ifdef CPL_DBGSW_WINSOCK_send
    CPL_WIN_OUTPUT_EX
      ( "send_cache(id:%d, length:%d, time:%d)\n"
      , backup_req_info->m_ID
      , backup_req_info->m_size
      , timeGetTime() - t_base
      );
#endif
  }

  //this->m_evt_send.ResetEvent();

  auto res = zzf_Refresh();
  if( res < 0 )
  {
    WIN::OutputWSALastErrorMessage();
    return SOCKET_ERROR;
  }

  return size_count;
}


// _____________________________________________________________________________
/// @brief
/// キャッシュにプッシュ。\n
/// キャッシュが満杯になるたび送信される。
/// 
/// @param[in,out] io_send_paren
///   : 送信パラメータ。SendParamへの参照を渡す。
/// 
/// @note 
/// 暗号化単位のセグメントの関係で、キャッシュを用意している。
/// winsockでもキャッシュがあるので、速度の面ではあまり効果はないと思われる。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
S64 SocketTCP::ThreadSend::zzf_SendQueue_Push( SendParam& io_send_param ) const
{
  auto locking = LockMutex( m_mutex_cache );

  U8_cptr      addr   = io_send_param.in_buf;
  S64          length = io_send_param.in_length;
  size_t const temp_buf_size = m_cache_buffer.GetReserveSize();

  //
  while( length > 0 )
  {
    // 追加サイズを計算。
    // ※ 追加サイズは最大でもキャッシュサイズなので、それほど大きなサイズにはならない。
    //    なので、size_t(32bit)でもOK.
    size_t add_size =
      (size_t)Min<S64>( length, temp_buf_size - m_cache_buffer.GetSize() );

    m_cache_buffer.AddEnd( addr, add_size ); // ※ キャッシュサイズは
    length -= add_size;

    if( m_cache_buffer.GetSize() >= temp_buf_size )
    {
      io_send_param.out_error = zzf_Refresh( io_send_param.out_size_sending );

      if( io_send_param.out_error )
      {
        return 0;
      }
    }
  }

  return io_send_param.in_length;
}


// _____________________________________________________________________________
/// @brief キャッシュバッファの送信。
/// 
/// @param[in] out_ptr_size : 送信されたサイズを受け取る変数のポインタ。
/// 
/// @return
/// 正常終了で0を返す。\n
/// 失敗した場合、エラーコードを返す。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
SDEF SocketTCP::ThreadSend::zzf_Refresh( volatile S64* out_ptr_size ) const
{
  auto locking = LockMutex( m_mutex_cache );

  using FMH = Socket_i_::FragmentHeader;
  const u_long header_size = sizeof(FMH);
  static U8 compress_buf[ (int)(CPL_WIN_WS_SOCKET_TEMP_BUF_SIZE * 1.001 + 12) ];

  auto org_size = m_cache_buffer.size();
  if( org_size <= header_size )
  {
    return 0;
  }

  //
  auto org_data_size = org_size - header_size;
  FMH* ptr_header = reinterpret_cast<FMH*>(&m_cache_buffer[0]);
  // 圧縮関数の呼び出し。
  u_long compress_size = (u_long)sizeof(compress_buf);
  (*WinSock::zlib_compress2())
    ( compress_buf
    , &compress_size
    , (u_char_cptr)(ptr_header+1)
    , (long)org_size-header_size
    , 9
    );

  // 圧縮サイズ判定（元のサイズより小さくなっていたら）
  if( compress_size < org_data_size )
  {
    ptr_header->f_compress = true;
    ptr_header->size       = (U16)compress_size;
    CPL_MEMCPY( ptr_header+1, compress_buf, compress_size ); // 圧縮データをコピー
  }
  else
  {
    ptr_header->f_compress = false;
    ptr_header->size       = (U16)org_data_size;
  }
  
  U32 data_size = ptr_header->size; // サイズを保存
  // 暗号化判定。
  cref<std::string> password = m_ptr_socket->GetSendPassword();
  bool flag_password = (!password.empty()  &&  password != "");
  if( flag_password )
  {
    Core::Encryption(
      &m_cache_buffer[ header_size ],
      data_size,
      &password[0],
      static_cast<int>(password.length()) );
  }

  S64 offs = 0,
      send_size = data_size + header_size;
  SendParam sp;
  while( send_size > 0 )
  {
    sp.in_buf    = &m_cache_buffer[ (size_t)offs ]; // ※ キャッシュはサイズが小さいので、size_t(32bit)でOK.
    sp.in_length = send_size;
    sp.out_size_sending = out_ptr_size;
    S64 ret = zzf_SendEx( sp );
    if( ret <= 0 )
    {
      return sp.out_error;
    }
    send_size -= ret;
    offs      += ret;
  }

  m_cache_buffer.Clear();
  FMH header;
  m_cache_buffer.AddEnd( (U8_cptr)&header, header_size );

  return 0;
}


  //struct A{};

  //template<class T>
  //struct X{
  //  explicit X( X const& ) = delete;
  //  explicit X( X&& ) = delete;
  //  explicit X( rvref<T> )=delete;

  //  explicit X( T ){}
  //  //X( A* p ){}

  //  //X& operator=(X const&) = delete;
  //  //X& operator = ( T ){ return *this; }
  //};

// _____________________________________________________________________________
/// @brief 送信スレッド。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
size_t SocketTCP::ThreadSend::operator()()
{
  m_flag_life_thread.store( true );

  typedef std::chrono::milliseconds time_msec;
  Core::TimeCounter counter;
  time_msec         extra_time( 3000 );
  time_msec         life_end( extra_time );

  while( !IsCancelThread() )
  {
    if( zzf_SendQueue() > 0 )
    {
      life_end = counter.GetTime_msec() + extra_time;
    }

    CPL_SleepNextTimeSlice();

    CPL_EXCTION_( m_mutex_request, m_mutex_data, m_mutex_cache )
    {
      time_msec ms = counter.GetTime_msec();
      if( ms > life_end )
      {
        m_flag_life_thread.store( false );
        return 0;
      }
    }
  }

  return 0;
}


// _____________________________________________________________________________
/// @brief バッファ追加。
/// 
/// @param[in] in_buffer_table[] : ByteBufferの配列。
/// @param[in] in_table_num      : 配列の個数。
/// 
/// @return
///   送信リクエストの識別IDを返す。\n
///   送信されない場合、-1を返す。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
U32
  SocketTCP::ThreadSend::AddData
    ( SpByteBuffer const in_buffer_table[]
    , int                in_table_num
    )
{
  // トータルサイズを求める。
  U32 buf_size = 0;
  CPL_LOOP2( in_table_num, idx )
  {
    if( in_buffer_table[ idx ] )
    {
      buf_size += static_cast<U32>(in_buffer_table[ idx ]->size());
    }
  }
  if( buf_size <= 0 )
  {
    return -1;
  }

  Core::String module_name;
  GetModuleName( module_name );
  str_type str[ 1024 ];
  CPL_SPRINTF
    ( str
    , SwS("%s: [P:%d][T:%d]: 送信キャッシュに追加(size:%d): AutoLockSendCacheAccess")
    , (str_cptr)module_name
    , (int)GetCurrentProcessId()
    , (int)Core::GetThreadID()
    , (int)buf_size
    );
  
  auto locking_req = LockMutex( m_mutex_request );

  // リクエスト情報を作成
  U32 header_size = RequestInfo::ComputeSize(in_table_num);
  RequestInfoBuffer req_block( header_size );
  req_block->InitSignature();
  req_block->m_ID   = id_counter()++;
  req_block->m_size = header_size + buf_size;
  req_block->m_segment_num = in_table_num;
  CPL_LOOP2( in_table_num, idx )
  {
    req_block->m_segment[ idx ].m_size = (U32)in_buffer_table[ idx ]->size();
  }
  // 
  m_send_request_queue.push_back( req_block );

  // バッファ
  auto locking_data = LockMutex( m_mutex_data );

  //if( in_flag_copy ) // バッファのコピーを作成するかどうかチェック。
  //{
  //  // キャッシュバッファキューに追加。
  //  CPL_LOOP2( in_table_num, idx )
  //  {
  //    // コピー用バッファ作成。
  //    U32 seg_size = static_cast<U32>(in_buffer_table[ idx ]->size());
  //    SpByteBuffer sp_temp_buf( new STLBuffer( seg_size ) );
  //    // コピー。
  //    memcpy( &(*sp_temp_buf)[0], &(*in_buffer_table[ idx ])[0], seg_size );
  //    // キューに追加。
  //    m_send_data_queue.push( sp_temp_buf );
  //  }
  //}
  //else
  //{
    // データバッファキューに追加。
    CPL_LOOP2( in_table_num, idx )
    {
      m_send_data_queue.Push( in_buffer_table[ idx ], false );
    }
  //}

  if( !IsThread() ) // スレッドが無い場合。
  {
    //Create(); // スレッドを作成する。
    Start(); // スレッドを開始する。
  }

  return req_block->m_ID;
}


// _____________________________________________________________________________
/// @brief 送信済みのセグメント数を取得。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
int SocketTCP::ThreadSend::GetCompleteSegmentCacheNum() const
{
  return m_complete_seg_num;
}


// _____________________________________________________________________________
/// @brief 指定した送信IDがリクエストキューに残っている間、ウエイトをかける。
/// 
/// @param[in] in_send_ID : 送信リクエストID.
/// @param[in] in_time_point_end : タイムアウトにするタイムポイントを指定。
/// 
/// @retval false : リクエストキューに指定されたIDは残っていない。
/// @retval true  : タイムアウト、もしくは強制終了。
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
bool
  SocketTCP::ThreadSend::WaitSendRequest
    ( _in<U32      > in_send_ID
    , _in<TimePoint> in_time_point_end
    ) const
{
  bool is_request_left = false; // リクエストが残っているかどうか（true=残っている）

  do {
    is_request_left = false;

    if( IsCancelThread() )
    {
      return true;
    }

    //CPL_LOOP_ITR( m_send_request_queue, itr_req )
    for( auto& buf_req : m_send_request_queue )
    {
      if( buf_req->m_ID == in_send_ID )
      {
        is_request_left = true;
        break;
      }
    }

    if( IsTimeout( in_time_point_end ) )
    {
      return is_request_left;
    }

    CPL_SleepNextTimeSlice();

  } while( is_request_left );

  return false;
}




} // namespace WIN
} // namespace CPL