#pragma once
/*
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
 */

/**
 * $Id: 2caa99839ef3f484e2bbcb3aec4800e2761ba1ab $
 *
 * @file io/channel.h
 * @brief 2-way channels based on kqueue and atomic queues.
 *
 * @copyright 2016 Alan DeKok (aland@freeradius.org)
 */
RCSIDH(channel_h, "$Id: 2caa99839ef3f484e2bbcb3aec4800e2761ba1ab $")

#ifdef __cplusplus
extern "C" {
#endif

#include <freeradius-devel/util/table.h>

/**
 *  A two-way channel (i.e. pipe) for exchanging information
 *
 *  While the channels are two-way, they are designed to have a
 *  "frontend" writing requests to the channel, and a "worker" reading
 *  requests, and writing replies back to the frontend.
 */
typedef struct fr_channel_s fr_channel_t;

/*
 *	Forward declaration until such time as we fix the code so that
 *	the network threads can push transports to worker threads.
 */
typedef struct fr_listen fr_listen_t;

#ifdef __cplusplus
}
#endif

#include <freeradius-devel/io/base.h>
#include <freeradius-devel/io/control.h>
#include <freeradius-devel/io/message.h>
#include <freeradius-devel/util/dlist.h>
#include <freeradius-devel/util/heap.h>
#include <freeradius-devel/util/log.h>

#include <sys/types.h>
#include <sys/event.h>

#ifdef __cplusplus
extern "C" {
#endif

typedef enum fr_channel_event_t {
	FR_CHANNEL_ERROR = 0,
	FR_CHANNEL_DATA_READY_RESPONDER,
	FR_CHANNEL_DATA_READY_REQUESTOR,
	FR_CHANNEL_OPEN,
	FR_CHANNEL_CLOSE,

	FR_CHANNEL_NOOP,
	FR_CHANNEL_EMPTY,
} fr_channel_event_t;

/** Statistics for the channel
 *
 */
typedef struct {
	uint64_t       		outstanding; 	//!< Number of outstanding requests with no reply.
	uint64_t		signals;	//!< Number of kevent signals we've sent.
	uint64_t		resignals;	//!< Number of signals resent.

	uint64_t		packets;	//!< Number of actual data packets.

	uint64_t		kevents;	//!< Number of times we've looked at kevents.

	fr_time_t		last_write;	//!< Last write to the channel.
	fr_time_t		last_read_other; //!< Last time we successfully read a message from the other the channel
	fr_time_delta_t		message_interval; //!< Interval between messages.

	fr_time_t		last_sent_signal; //!< The last time when we signaled the other end.
} fr_channel_stats_t;


/**
 *  Channel information which is added to a message.
 *
 *  The messages are just for exchanging packet data.  The channel
 *  data structure is for exchanging requests and replies.
 */
typedef struct {
	fr_message_t		m;		//!< the message header

	union {
		/*
		 *	Messages have a sequence number / ack while
		 *	they're in a channel.
		 */
		struct {
			uint64_t		sequence;	//!< sequence number
			uint64_t		ack;		//!< ACK of the sequence number from the other end
		} live;

		/*
		 *	Once messages are pulled out of a channel by
		 *	the scheduler, we need to cache the channel
		 *	somewhere.  So we cache it in fields which are now unused.
		 */
		struct {
			fr_channel_t		*ch;		//!< channel where this messages was received
			fr_heap_index_t		heap_id;	//!< for the various queues
		} channel;
	};

	union {
		struct {
			fr_time_t		recv_time;	//!< time original request was received (network -> worker)
		} request;

		struct {
			fr_time_delta_t		cpu_time;		//!< Total CPU time, including predicted work, (only worker -> network).
			fr_time_delta_t		processing_time; 	//!< Actual processing time for this packet (only worker -> network).
			fr_time_t		request_time;		//!< Timestamp of the request packet.
	        } reply;
	};

	uint32_t	priority;				//!< Priority of this packet.

	void		*packet_ctx;				//!< Packet specific context for holding client
								//!< information, and other proto_* specific information
								//!< that needs to be passed to the request.

	fr_listen_t	*listen;				//!< for tracking packet transport, etc.
} fr_channel_data_t;

#define PRIORITY_NOW    (1 << 16)
#define PRIORITY_HIGH   (1 << 15)
#define PRIORITY_NORMAL (1 << 14)
#define PRIORITY_LOW    (1 << 13)

extern fr_table_num_sorted_t const channel_signals[];
extern size_t channel_signals_len;
extern fr_table_num_sorted_t const channel_packet_priority[];
extern size_t channel_packet_priority_len;

fr_channel_t *fr_channel_create(TALLOC_CTX *ctx, fr_control_t *frontend, fr_control_t *worker, bool same) CC_HINT(nonnull);

int	fr_channel_send_request(fr_channel_t *ch, fr_channel_data_t *cm) CC_HINT(nonnull);
bool	fr_channel_recv_request(fr_channel_t *ch) CC_HINT(nonnull);

int	fr_channel_send_reply(fr_channel_t *ch, fr_channel_data_t *cd) CC_HINT(nonnull);
int	fr_channel_null_reply(fr_channel_t *ch) CC_HINT(nonnull);

bool	fr_channel_recv_reply(fr_channel_t *ch) CC_HINT(nonnull);

typedef void (*fr_channel_recv_callback_t)(void *ctx, fr_channel_t *ch, fr_channel_data_t *cd);
int	fr_channel_set_recv_reply(fr_channel_t *ch, void *ctx, fr_channel_recv_callback_t recv_reply) CC_HINT(nonnull(1,3));
int	fr_channel_set_recv_request(fr_channel_t *ch, void *ctx, fr_channel_recv_callback_t recv_reply) CC_HINT(nonnull(1,3));

int	fr_channel_responder_sleeping(fr_channel_t *ch) CC_HINT(nonnull);

int	fr_channel_service_kevent(fr_channel_t *ch, fr_control_t *c, struct kevent const *kev) CC_HINT(nonnull);
fr_channel_event_t	fr_channel_service_message(fr_time_t when, fr_channel_t **p_channel, void const *data, size_t data_size) CC_HINT(nonnull);

bool	fr_channel_active(fr_channel_t *ch) CC_HINT(nonnull);

int	fr_channel_signal_open(fr_channel_t *ch) CC_HINT(nonnull);

int	fr_channel_signal_responder_close(fr_channel_t *ch) CC_HINT(nonnull);
int	fr_channel_responder_ack_close(fr_channel_t *ch) CC_HINT(nonnull);

void	fr_channel_responder_uctx_add(fr_channel_t *ch, void *ctx) CC_HINT(nonnull);
void	*fr_channel_responder_uctx_get(fr_channel_t *ch) CC_HINT(nonnull);
void	fr_channel_requestor_uctx_add(fr_channel_t *ch, void *ctx) CC_HINT(nonnull);
void	*fr_channel_requestor_uctx_get(fr_channel_t *ch) CC_HINT(nonnull);


void	fr_channel_stats_log(fr_channel_t const *ch, fr_log_t const *log, char const *file, int line);

#ifdef __cplusplus
}
#endif
