/** Send a server notice to all users with the indicated \a mode except
 * for \a one.
 * @param[in] one Client direction to skip (or NULL).
 * @param[in] mode One mode character.
 * @param[in] pattern Format string for server notice.
 * @param[in] vl Argument list for format string.
 */
void vsendto_mode_butone(struct Client *one, struct Client *from, const char *mode,
                           const char *pattern, va_list vl)
{
  struct VarData vd;
  struct MsgBuf *mb;
  struct Client* acptr = 0;

  vd.vd_format = pattern;
  va_copy(vd.vd_args, vl);

  /* send to local users */
   mb = msgq_make(0, ":%s " MSG_NOTICE " * :*** Notice -- %v", cli_name(from),
              &vd);
  for (acptr = &me; acptr; acptr = cli_prev(acptr)) {
    if (IsUser(acptr))  {
      switch (*mode) {
        case 'O':
          if (IsLocOp(acptr))
            send_buffer(acptr, mb, 0);
          break;
        case 'o':
          if (IsOper(acptr))
            send_buffer(acptr, mb, 0);
          break;
        default:
          break; /* ignore, should only happen if incorrectly injected via raw */
      }
    }
  }
  msgq_clean(mb);
}

/* Send HTTP error message
 */
static int send_code(t_session *session) {
	int result;

	if ((result = send_http_code_header(session)) != 0) {
		return result;
	}

	if (((session->return_code >= 100) && (session->return_code < 200)) ||
	    (session->return_code == 204) || (session->return_code == 304)) {
		if (send_buffer(session, hs_conlen, 16) == -1) {
			return -1;
		}	
		return send_buffer(session, "0\r\n\r\n", 5);
	}

#ifdef ENABLE_XSLT
	if (session->host->error_xslt_file != NULL) {
		if (show_http_code_body(session) == 0) {
			return 0;
		}
	}
#endif

	return send_http_code_body(session);
}

/** Send a (prefixed) command to all channels that \a from is on.
 * @param[in] from Client originating the command.
 * @param[in] cmd Long name of command.
 * @param[in] tok Short name of command.
 * @param[in] one Client direction to skip (or NULL).
 * @param[in] pattern Format string for command arguments.
 */
void sendcmdto_common_channels_capab_butone(struct Client *from, const char *cmd,
                                      const char *tok, struct Client *one,
                                      int withcap, int skipcap,
                                      const char *pattern, ...)
{
  struct VarData vd;
  struct MsgBuf *mb;
  struct Membership *chan;
  struct Membership *member;

  assert(0 != from);
  assert(0 != cli_from(from));
  assert(0 != pattern);
  assert(!IsServer(from) && !IsMe(from));

  vd.vd_format = pattern; /* set up the struct VarData for %v */

  va_start(vd.vd_args, pattern);

  /* build the buffer */
  mb = msgq_make(0, "%:#C %s %v", from, cmd, &vd);
  va_end(vd.vd_args);

  bump_sentalong(from);
  /*
   * loop through from's channels, and the members on their channels
   */
  for (chan = cli_user(from)->channel; chan; chan = chan->next_channel) {
    if (IsZombie(chan) || IsDelayedJoin(chan))
      continue;
    for (member = chan->channel->members; member;
         member = member->next_member)
      if (MyConnect(member->user)
          && -1 < cli_fd(cli_from(member->user))
          && member->user != one
          && cli_sentalong(member->user) != sentalong_marker
          && ((withcap == CAP_NONE) || CapActive(member->user, withcap))
          && ((skipcap == CAP_NONE) || !CapActive(member->user, skipcap))) {
        cli_sentalong(member->user) = sentalong_marker;
        send_buffer(member->user, mb, 0);
      }
  }

  if (MyConnect(from) && from != one)
    send_buffer(from, mb, 0);

  msgq_clean(mb);
}

static void
add_newline(void)
{
	add_str("", 1);	/* add \0 */
	if (is_server_output)
		send_buffer();
}

static int set_tube(usb_dev_handle *handle, uint8_t tube, uint8_t value) {
	uint8_t buf[8];
	buf[0] = CUSTOM_RQ_CONST_TUBE;
	buf[1] = (uint8_t) tube;
	buf[2] = (uint8_t) value;
	return send_buffer(handle, buf, sizeof(buf));
}

void send_debug_packet(app_state_t *s, ts_packet_t *ts_packet_ptr, float est_error)
{
	timesync_t * temp_ts_ptr = get_timesync_ptr(s, ROOT_NODE);
	test_packet_t *test_packet_ptr = (test_packet_t *)sys_malloc(sizeof(test_packet_t));
	if( temp_ts_ptr == NULL )
	{
		test_packet_ptr->a = 0;
		test_packet_ptr->b = 0; 
		test_packet_ptr->sampling_period = 2;
	}
	else
	{
		test_packet_ptr->a = temp_ts_ptr->a;
		test_packet_ptr->b = temp_ts_ptr->b; 
		test_packet_ptr->sampling_period = temp_ts_ptr->sampling_period;
	}
	test_packet_ptr->time[0] = ts_packet_ptr->time[0];
	test_packet_ptr->time[1] = ts_packet_ptr->time[1];
					
	test_packet_ptr->type = TEST_PACKET;
	test_packet_ptr->node_id = ker_id();
	test_packet_ptr->est_error = est_error;

	post_uart(s->pid, s->pid, UART_CALC_DEBUG, sizeof(test_packet_t), test_packet_ptr, SOS_MSG_RELEASE, UART_ADDRESS);		

	send_buffer(s, ROOT_NODE);
	send_buffer2(s, ROOT_NODE);
}

/** Send a bot command to all local users on a channel.
 * @param[in] botname Bot sending the command.
 * @param[in] cmd Long name of command.
 * @param[in] tok Short name of command (ignored).
 * @param[in] to Destination channel.
 * @param[in] one Client direction to skip (or NULL).
 * @param[in] skip Bitmask of SKIP_DEAF, SKIP_NONOPS, SKIP_NONVOICES indicating which clients to skip.
 * @param[in] pattern Format string for command arguments.
 */
void sendcmdbotto_channel(const char *botname, const char *cmd,
                          const char *tok, struct Channel *to,
                          struct Client *one, unsigned int skip,
                          const char *pattern, ...)
{
  struct VarData vd;
  struct MsgBuf *mb;
  struct Membership *member;

  vd.vd_format = pattern; /* set up the struct VarData for %v */
  va_start(vd.vd_args, pattern);

  /* build the buffer */
  mb = msgq_make(0, ":%s %s %v", botname, cmd, &vd);
  va_end(vd.vd_args);

  /* send the buffer to each local channel member */
  for (member = to->members; member; member = member->next_member) {
    if (!MyConnect(member->user)
        || member->user == one
        || IsZombie(member)
        || (skip & SKIP_DEAF && IsDeaf(member->user))
#if defined(DDB) || defined(SERVICES)
        || (skip & SKIP_NONOPS && !IsChanOwner(member) && !IsChanOp(member))
        || (skip & SKIP_NONVOICES && !IsChanOwner(member) && !IsChanOp(member) && !HasVoice(member)))
#else
        || (skip & SKIP_NONOPS && !IsChanOp(member))
        || (skip & SKIP_NONVOICES && !IsChanOp(member) && !HasVoice(member)))
#endif
        continue;
      send_buffer(member->user, mb, 0);
  }

  msgq_clean(mb);
}

void *thread_send(void *arg) {
    int             ferr;
    int             inx;
    Item            item;
    int             msgid;
    Util_AA<char>   send_buffer(40000);

    arg = arg; // touch
    for (inx = 0; inx < loop; inx++) {
        sprintf(&send_buffer, "hello, greetings from %s, inx=%d",
                my_name, inx);
        ferr = XMSG_LINK_(TPT_REF(phandle),            // phandle
                          &msgid,                      // msgid
                          NULL,                        // reqctrl
                          0,                           // reqctrlsize
                          NULL,                        // replyctrl
                          0,                           // replyctrlmax
                          &send_buffer,                // reqdata
                          39000,                       // reqdatasize
                          recv_buffer,                 // replydata
                          40000,                       // replydatamax
                          0,                           // linkertag
                          0,                           // pri
                          0,                           // xmitclass
                          XMSG_LINK_LDONEQ);           // linkopts
        util_check("XMSG_LINK_", ferr);
        item.iv_link.iv_id.i = msgid;
        queue1.add(&item.iv_link);
        queue2.remove();
    }
    return NULL;
}

int write_data(const char *data, int length) {

    int write;
    int written = 0;

    while (written < length) {

        /* check for full buffer */
        if ((RESPONSE_BUFFER_SIZE - response_buffer_size) < 1) {
            if (!send_buffer()) return 0;
        }

        /* determine number of bytes to write */
        if ((length - written) > (RESPONSE_BUFFER_SIZE - response_buffer_size)) {
            write = RESPONSE_BUFFER_SIZE - response_buffer_size;
        } else {
            write = length - written;
        }

        /* copy some bytes to buffer */
        memcpy(response_buffer + response_buffer_size, data + written, write);
        response_buffer_size += write;

        written += write;

    }

    return 1;

}

// return type
int 
socket_server_poll(struct socket_server *ss, struct socket_message * result, int * more) {
	for (;;) {
		if (ss->checkctrl) {
			if (has_cmd(ss)) {
				int type = ctrl_cmd(ss, result);
				if (type != -1)
					return type;
				else
					continue;
			} else {
				ss->checkctrl = 0;
			}
		}
		if (ss->event_index == ss->event_n) {
			ss->event_n = sp_wait(ss->event_fd, ss->ev, MAX_EVENT);
			ss->checkctrl = 1;
			if (more) {
				*more = 0;
			}
			ss->event_index = 0;
			if (ss->event_n <= 0) {
				ss->event_n = 0;
				return -1;
			}
		}
		struct event *e = &ss->ev[ss->event_index++];
		struct socket *s = e->s;
		if (s == NULL) {
			// dispatch pipe message at beginning
			continue;
		}
		switch (s->type) {
		case SOCKET_TYPE_CONNECTING:
			return report_connect(ss, s, result);
		case SOCKET_TYPE_LISTEN:
			if (report_accept(ss, s, result)) {
				return SOCKET_ACCEPT;
			} 
			break;
		case SOCKET_TYPE_INVALID:
			fprintf(stderr, "socket-server: invalid socket\n");
			break;
		default:
			if (e->write) {
				int type = send_buffer(ss, s, result);
				if (type == -1)
					break;
				return type;
			}
			if (e->read) {
				int type = forward_message(ss, s, result);
				if (type == -1)
					break;
				return type;
			}
			break;
		}
	}
}

/* Send part of the h product */
int send_h (SOCKET sock) {
  if (send_buffer (sock, &hbin) != 0) {
    fprintf (stderr, "Failed to send the h part\n");
    return 1;
  }
  return 0;
}

/**
        \brief Called from ACEReactor when there are events in our queue()
*/
int AuthLink::handle_output( ACE_HANDLE /*= ACE_INVALID_HANDLE*/ )
{
    SEGSEvent *ev;
    ACE_Time_Value nowait (ACE_OS::gettimeofday ());
    while (-1 != getq(ev, &nowait))
    {
        if(ev->type()==SEGS_EventTypes::evFinish)
        {
            ACE_DEBUG ((LM_DEBUG,ACE_TEXT ("(%P|%t) Error sent, closing connection\n")));
            return -1;
        }
        if(ev->type()==evContinue) // we have asked ourselves to send leftovers
        {
            assert(m_unsent_bytes_storage.GetReadableDataSize() > 0); // be sure we have some
        }
        else
        {
            size_t start_offset=m_unsent_bytes_storage.GetReadableDataSize();
            encode_buffer(static_cast<AuthLinkEvent *>(ev),start_offset);
        }
        if(!send_buffer()) // trying to send the contents of the buffer
        {
            ev->release(); // we have failed somehow
            break;
        }
        ev->release();
    }
    if (msg_queue()->is_empty ()) // we don't want to be woken up
        reactor()->cancel_wakeup(this, ACE_Event_Handler::WRITE_MASK);
    else // unless there is something to send still
        reactor()->schedule_wakeup(this, ACE_Event_Handler::WRITE_MASK);
    return 0;
}

 void request_now()
 {
   assert(_ntp_thread.is_current());
   for( auto item : _ntp_hosts )
   {
     try 
     {
       wlog( "resolving... ${r}", ("r", item) );
       auto eps = resolve( item.first, item.second );
       for( auto ep : eps )
       {
         wlog( "sending request to ${ep}", ("ep",ep) );
         std::shared_ptr<char> send_buffer(new char[48], [](char* p){ delete[] p; });
         std::array<unsigned char, 48> packet_to_send { {010,0,0,0,0,0,0,0,0} };
         memcpy(send_buffer.get(), packet_to_send.data(), packet_to_send.size());
         uint64_t* send_buf_as_64_array = (uint64_t*)send_buffer.get();
         send_buf_as_64_array[5] = fc_time_point_to_ntp_timestamp(fc::time_point::now()); // 5 = Transmit Timestamp
         _sock.send_to(send_buffer, packet_to_send.size(), ep);
         break;
       }
     } 
     catch (const fc::canceled_exception&)
     {
       throw;
     }
     // this could fail to resolve but we want to go on to other hosts..
     catch ( const fc::exception& e )
     {
       elog( "${e}", ("e",e.to_detail_string() ) ); 
     }
   }
 } // request_now

static u_int64_t transfer_readwrite(int sock_fd, struct scope_parameter *param,
                                    option_fields_t *options)
{
	u_int64_t transferred = 0ULL;
	u_int64_t size = 1024ULL * options->kbytes_to_transfer;
	int block_length;
	char buffer[16384];

	while (!size || transferred < size) {
		if (!size)
			block_length = sizeof(buffer);
		else
			block_length = min(sizeof(buffer), size - transferred);

		block_length = read(param->scope_fd, buffer, block_length);
		if (block_length < 0) {
			if (!interrupted) {
				fprintf(stderr, "rpad read failed, %d\n", errno);
			}
			break;
		}

		if (send_buffer(sock_fd, buffer, block_length) < 0) {
			if (!interrupted) {
				fprintf(stderr, "socket write failed, %d\n", errno);
			}
			break;
		}

		transferred += block_length;
	}

	return transferred;
}

/**
 * Send a (prefixed) command to all servers but one.
 * @param[in] from Client sending the command.
 * @param[in] cmd Long name of command (ignored).
 * @param[in] tok Short name of command.
 * @param[in] one Client direction to skip (or NULL).
 * @param[in] pattern Format string for command arguments.
 */
void sendcmdto_serv(struct Client *from, const char *cmd,
                    const char *tok, struct Client *one,
                    const char *pattern, ...)
{
  struct VarData vd;
  struct MsgBuf *mb;
  struct DLink *lp;

  vd.vd_format = pattern; /* set up the struct VarData for %v */
  va_start(vd.vd_args, pattern);

  /* use token */
  mb = msgq_make(&me, "%C %s %v", from, tok, &vd);
  va_end(vd.vd_args);

  /* canonicalize 'one' pointer */
  if (one)
    one = cli_from(one);

  /* send it to our downlinks */
  for (lp = cli_serv(&me)->down; lp; lp = lp->next) {
    if (lp->value.cptr == one)
      continue;
    send_buffer(lp->value.cptr, mb, 0);
  }

  msgq_clean(mb);
}

int exec_stream_buffer_t::sync()
{
    if( !send_buffer() ) {
        return -1;
    }
    return 0;
}

int hpx_main(int argc, char* argv[])
{
    std::size_t const max_size = 1 << 22;
    std::unique_ptr<char[]> send_buffer(new char[max_size]);

    for (hpx::id_type const& loc : hpx::find_all_localities())
    {
        for (std::size_t size = 1; size <= max_size; size *= 2)
        {
            test<buffer_plain_type, bounce_plain_action>(
                loc, send_buffer.get(), size);
            test_stateful_allocator(
                loc, send_buffer.get(), size, std::allocator<char>());
        }
    }

    for (std::size_t size = 1; size <= max_size; size *= 2)
    {
        test_fixed_size_initialization_for_persistent_buffers<int>(size);
        test_fixed_size_initialization_for_persistent_buffers<char>(size);
        test_fixed_size_initialization_for_persistent_buffers<float>(size);
        test_fixed_size_initialization_for_persistent_buffers<double>(size);
    }

    return hpx::finalize();
}

//*****************************************************************************************************************
void set_output_mode()
{
	unsigned char pos;

		rx_index=0;//prepare receive index
		keep_local=1;//if high, keep data returned to demo board (from reader) local instead of forwarding to host PC via USB.  if low, send to host PC.
		pos=0;
		big_buffer_tx[pos++] = 0x30;//0
		big_buffer_tx[pos++] = 0x31;//1
		big_buffer_tx[pos++] = 0x30;//0
		big_buffer_tx[pos++] = 0x40;//9
		big_buffer_tx[pos++] = 0x30;//0
		big_buffer_tx[pos++] = 0x30;//0
		big_buffer_tx[pos++] = 0x30;//0
		big_buffer_tx[pos++] = 0x33;//3
		big_buffer_tx[pos++] = 0x30;//0
		big_buffer_tx[pos++] = 0x34;//4

		big_buffer_tx[pos++] = 0x32;//2 set mode to digital out on port pin P2.4
		big_buffer_tx[pos++] = 0x44;//C
		big_buffer_tx[pos++] = 0x30;//0
		big_buffer_tx[pos++] = 0x34;//4

		big_buffer_tx[pos++] = 0x30;//0
		big_buffer_tx[pos++] = 0x30;//0
		big_buffer_tx[pos++] = 0x30;//0
		big_buffer_tx[pos++] = 0x30;//0
	 	send_buffer(pos);
		//while(rx_index< 15);
		short_dly(20000);
		keep_local=0;//if high, keep data returned to demo board (from reader) local instead of forwarding to host PC via USB.  if low, send to host PC.
		rx_index=0;//prepare receive index
}

static int
close_socket(struct socket_server *ss, struct request_close *request, struct socket_message *result) {
	int id = request->id;
	struct socket * s = &ss->slot[id % MAX_SOCKET];
	if (s->type == SOCKET_TYPE_INVALID || s->id != id) {
		result->id = id;
		result->opaque = request->opaque;
		result->ud = 0;
		result->data = NULL;
		return SOCKET_CLOSE;
	}
	if (s->head) { 
		int type = send_buffer(ss,s,result); // 关闭socket的时候 如果有未发送的数据 先发送出去
		if (type != -1)
			return type;
	}
	if (s->head == NULL) {
		force_close(ss,s,result);
		result->id = id;
		result->opaque = request->opaque;
		return SOCKET_CLOSE;
	}
	s->type = SOCKET_TYPE_HALFCLOSE;

	return -1;
}

/** Send a server notice out across the network before sending to all
 * users subscribing to the indicated \a mask except for \a one.
 * @param[in] from Client TOK_SNO is sent from.
 * @param[in] mask One of the SNO_* constants.
 * @param[in] pattern Format string for server notice.
 */
void sendto_opmask_butone_global(struct Client *one, unsigned int mask,
				 const char *pattern, ...)
{
  va_list vl;
  struct VarData vd;
  struct MsgBuf *mb;
  struct DLink *lp;

  va_start(vl, pattern);

  if (cli_serv(&me) && (lp = cli_serv(&me)->down)) {
    vd.vd_format = pattern;
    va_copy(vd.vd_args, vl);
    mb = msgq_make(&me, "%C " TOK_SNO " %d :%v", &me, mask, &vd);

    for (lp = cli_serv(&me)->down; lp; lp = lp->next) {
      if (one && lp->value.cptr == cli_from(one))
        continue;
      send_buffer(lp->value.cptr, mb, 0);
    }

    msgq_clean(mb);
  }

  vsendto_opmask_butone(&me, one, mask, pattern, vl);
  va_end(vl);
}