text: 
  /*
   * Odyssey.
   *
   * Scalable PostgreSQL connection pooler.
   */
  
  #include <kiwi.h>
  #include <machinarium.h>
  #include <odyssey.h>
  
  enum { OD_LKILL_CLIENT,
         OD_LRELOAD,
         OD_LSHOW,
         OD_LSTATS,
         OD_LSERVERS,
         OD_LCLIENTS,
         OD_LLISTS,
         OD_LSET,
         OD_LCREATE,
         OD_LDROP,
         OD_LPOOLS,
         OD_LPOOLS_EXTENDED,
         OD_LDATABASES,
         OD_LMODULE,
         OD_LERRORS,
         OD_LERRORS_PER_ROUTE,
         OD_LFRONTEND,
         OD_LROUTER,
         OD_LVERSION,
  };
  
  static od_keyword_t od_console_keywords[] = {
  	od_keyword("kill_client", OD_LKILL_CLIENT),
  	od_keyword("reload", OD_LRELOAD),
  	od_keyword("show", OD_LSHOW),
  	od_keyword("stats", OD_LSTATS),
  	od_keyword("servers", OD_LSERVERS),
  	od_keyword("clients", OD_LCLIENTS),
  	od_keyword("lists", OD_LLISTS),
  	od_keyword("set", OD_LSET),
  	od_keyword("pools", OD_LPOOLS),
  	od_keyword("pools_extended", OD_LPOOLS_EXTENDED),
  	od_keyword("databases", OD_LDATABASES),
  	od_keyword("create", OD_LCREATE),
  	od_keyword("module", OD_LMODULE),
  	od_keyword("errors", OD_LERRORS),
  	od_keyword("errors_per_route", OD_LERRORS_PER_ROUTE),
  	od_keyword("frontend", OD_LFRONTEND),
  	od_keyword("router", OD_LROUTER),
  	od_keyword("drop", OD_LDROP),
  	od_keyword("version", OD_LVERSION),
  	{ 0, 0, 0 }
  };
  
  static inline int od_console_show_stats_add(machine_msg_t *stream,
  					    char *database, int database_len,
  					    od_stat_t *total, od_stat_t *avg)
  {
  	assert(stream);
  	int offset;
  	machine_msg_t *msg;
  	msg = kiwi_be_write_data_row(stream, &offset);
  	if (msg == NULL)
  		return NOT_OK_RESPONSE;
  	int rc;
  	rc = kiwi_be_write_data_row_add(stream, offset, database, database_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	char data[64];
  	int data_len;
  	/* total_xact_count */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, total->count_tx);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* total_query_count */
  	data_len =
  		od_snprintf(data, sizeof(data), "%" PRIu64, total->count_query);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* total_received */
  	data_len =
  		od_snprintf(data, sizeof(data), "%" PRIu64, total->recv_client);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* total_sent */
  	data_len =
  		od_snprintf(data, sizeof(data), "%" PRIu64, total->recv_server);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* total_xact_time */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, total->tx_time);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* total_query_time */
  	data_len =
  		od_snprintf(data, sizeof(data), "%" PRIu64, total->query_time);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* total_wait_time */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, 0UL);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* avg_xact_count */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, avg->count_tx);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* avg_query_count */
  	data_len =
  		od_snprintf(data, sizeof(data), "%" PRIu64, avg->count_query);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* avg_recv */
  	data_len =
  		od_snprintf(data, sizeof(data), "%" PRIu64, avg->recv_client);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* avg_sent */
  	data_len =
  		od_snprintf(data, sizeof(data), "%" PRIu64, avg->recv_server);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* avg_xact_time */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, avg->tx_time);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* avg_query_time */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, avg->query_time);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* avg_wait_time */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, 0UL);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	return 0;
  }
  
  static inline od_retcode_t
  od_console_show_frontend_stats_err_add(machine_msg_t *stream,
  				       od_route_pool_t *route_pool)
  {
  	assert(stream);
  
  	for (size_t i = 0; i < OD_FRONTEND_STATUS_ERRORS_TYPES_COUNT; ++i) {
  		int offset;
  		int rc;
  		machine_msg_t *msg;
  
  		msg = kiwi_be_write_data_row(stream, &offset);
  		if (msg == NULL)
  			return NOT_OK_RESPONSE;
  
  		size_t total_count = od_err_logger_get_aggr_errors_count(
  			route_pool->err_logger, od_frontend_status_errs[i]);
  
  		char *err_type =
  			od_frontend_status_to_str(od_frontend_status_errs[i]);
  
  		rc = kiwi_be_write_data_row_add(stream, offset, err_type,
  						strlen(err_type));
  		if (rc != OK_RESPONSE) {
  			return rc;
  		}
  		char data[64];
  		int data_len;
  		/* error_type */
  		data_len = od_snprintf(data, sizeof(data), "%" PRIu64,
  				       total_count);
  
  		rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  		if (rc != OK_RESPONSE) {
  			return rc;
  		}
  	}
  
  	return OK_RESPONSE;
  }
  
  static inline int
  od_console_show_router_stats_err_add(machine_msg_t *stream,
  				     od_error_logger_t *err_logger)
  {
  	assert(stream);
  
  	for (size_t i = 0; i < OD_ROUTER_STATUS_ERRORS_TYPES_COUNT; ++i) {
  		int offset;
  		int rc;
  		machine_msg_t *msg;
  
  		msg = kiwi_be_write_data_row(stream, &offset);
  		if (msg == NULL) {
  			return NOT_OK_RESPONSE;
  		}
  
  		char *err_type =
  			od_router_status_to_str(od_router_status_errs[i]);
  
  		rc = kiwi_be_write_data_row_add(stream, offset, err_type,
  						strlen(err_type));
  		if (rc != OK_RESPONSE) {
  			return rc;
  		}
  
  		/* error_type */
  		char data[64];
  		int data_len;
  		data_len = od_snprintf(data, sizeof(data), "%" PRIu64,
  				       od_err_logger_get_aggr_errors_count(
  					       err_logger,
  					       od_router_status_errs[i]));
  
  		rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  		if (rc != OK_RESPONSE) {
  			return rc;
  		}
  	}
  
  	return OK_RESPONSE;
  }
  
  static int od_console_show_stats_cb(char *database, int database_len,
  				    od_stat_t *total, od_stat_t *avg,
  				    void **argv)
  {
  	machine_msg_t *stream = argv[0];
  	return od_console_show_stats_add(stream, database, database_len, total,
  					 avg);
  }
  
  static int od_console_show_err_frontend_stats_cb(od_route_pool_t *pool,
  						 void **argv)
  {
  	machine_msg_t *stream = argv[0];
  	return od_console_show_frontend_stats_err_add(stream, pool);
  }
  
  static int od_console_show_err_router_stats_cb(od_error_logger_t *l,
  					       void **argv)
  {
  	machine_msg_t *stream = argv[0];
  	return od_console_show_router_stats_err_add(stream, l);
  }
  
  static inline int od_console_show_stats(od_client_t *client,
  					machine_msg_t *stream)
  {
  	assert(stream);
  	od_router_t *router = client->global->router;
  	od_cron_t *cron = client->global->cron;
  
  	if (kiwi_be_write_row_descriptionf(
  		    stream, "sllllllllllllll", "database", "total_xact_count",
  		    "total_query_count", "total_received", "total_sent",
  		    "total_xact_time", "total_query_time", "total_wait_time",
  		    "avg_xact_count", "avg_query_count", "avg_recv", "avg_sent",
  		    "avg_xact_time", "avg_query_time",
  		    "avg_wait_time") == NULL) {
  		return NOT_OK_RESPONSE;
  	}
  
  	void *argv[] = { stream };
  	od_route_pool_stat_database(&router->route_pool,
  				    od_console_show_stats_cb,
  				    cron->stat_time_us, argv);
  
  	int rc = kiwi_be_write_complete(stream, "SHOW", 5);
  	if (rc == NOT_OK_RESPONSE) {
  		return rc;
  	}
  
  	return rc;
  }
  
  static inline od_retcode_t od_console_show_errors(od_client_t *client,
  						  machine_msg_t *stream)
  {
  	assert(stream);
  	od_router_t *router = client->global->router;
  
  	void *argv[] = { stream };
  
  	if (kiwi_be_write_row_descriptionf(stream, "sl", "error_type",
  					   "count") == NULL) {
  		return NOT_OK_RESPONSE;
  	}
  
  	int rc;
  	rc = od_route_pool_stat_err_router(
  		router, od_console_show_err_router_stats_cb, argv);
  
  	if (rc != OK_RESPONSE)
  		return rc;
  
  	rc = od_route_pool_stat_err_frontend(
  		&router->route_pool, od_console_show_err_frontend_stats_cb,
  		argv);
  
  	if (rc != OK_RESPONSE)
  		return rc;
  
  	rc = kiwi_be_write_complete(stream, "SHOW", 5);
  	return rc;
  }
  
  static inline int od_console_show_errors_per_route_cb(od_route_t *route,
  						      void **argv)
  {
  	machine_msg_t *stream = argv[0];
  	assert(stream);
  
  	if (!route || !route->extra_logging_enabled) {
  		return OK_RESPONSE;
  	}
  
  	for (size_t i = 0; i < OD_FRONTEND_STATUS_ERRORS_TYPES_COUNT; ++i) {
  		int offset;
  		int rc;
  		machine_msg_t *msg;
  		msg = kiwi_be_write_data_row(stream, &offset);
  		if (msg == NULL) {
  			/* message was not successfully allocated */
  			return NOT_OK_RESPONSE;
  		}
  
  		size_t total_count = od_err_logger_get_aggr_errors_count(
  			route->err_logger, od_frontend_status_errs[i]);
  
  		char *err_type =
  			od_frontend_status_to_str(od_frontend_status_errs[i]);
  
  		rc = kiwi_be_write_data_row_add(stream, offset, err_type,
  						strlen(err_type));
  
  		if (rc != OK_RESPONSE) {
  			return rc;
  		}
  
  		/* route user */
  
  		rc = kiwi_be_write_data_row_add(stream, offset,
  						route->rule->user_name,
  						strlen(route->rule->user_name));
  		if (rc != OK_RESPONSE) {
  			return rc;
  		}
  
  		/* route database */
  
  		rc = kiwi_be_write_data_row_add(stream, offset,
  						route->rule->db_name,
  						strlen(route->rule->db_name));
  		if (rc != OK_RESPONSE) {
  			return rc;
  		}
  
  		/* error_type */
  
  		char data[64];
  		int data_len;
  		data_len = od_snprintf(data, sizeof(data), "%" PRIu64,
  				       total_count);
  
  		rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  		if (rc != OK_RESPONSE) {
  			return rc;
  		}
  	}
  
  	for (size_t i = 0; i < OD_ROUTER_ROUTE_STATUS_ERRORS_TYPES_COUNT; ++i) {
  		int offset;
  		int rc;
  		machine_msg_t *msg;
  		msg = kiwi_be_write_data_row(stream, &offset);
  		if (msg == NULL)
  			return NOT_OK_RESPONSE;
  
  		size_t total_count = od_err_logger_get_aggr_errors_count(
  			route->err_logger, od_router_route_status_errs[i]);
  
  		char *err_type =
  			od_router_status_to_str(od_router_route_status_errs[i]);
  
  		rc = kiwi_be_write_data_row_add(stream, offset, err_type,
  						strlen(err_type));
  		if (rc != OK_RESPONSE) {
  			return rc;
  		}
  
  		/* route user */
  
  		rc = kiwi_be_write_data_row_add(stream, offset,
  						route->rule->user_name,
  						strlen(route->rule->user_name));
  		if (rc != OK_RESPONSE) {
  			return rc;
  		}
  
  		/* route database */
  
  		rc = kiwi_be_write_data_row_add(stream, offset,
  						route->rule->db_name,
  						strlen(route->rule->db_name));
  		if (rc != OK_RESPONSE) {
  			return rc;
  		}
  
  		/* error_type */
  
  		char data[64];
  		int data_len;
  		data_len = od_snprintf(data, sizeof(data), "%" PRIu64,
  				       total_count);
  
  		rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  		if (rc != OK_RESPONSE) {
  			return rc;
  		}
  	}
  
  	return OK_RESPONSE;
  }
  
  static inline od_retcode_t
  od_console_show_errors_per_route(od_client_t *client, machine_msg_t *stream)
  {
  	assert(stream);
  	od_router_t *router = client->global->router;
  
  	void *argv[] = { stream };
  
  	if (kiwi_be_write_row_descriptionf(stream, "sssl", "error_type", "user",
  					   "database", "count") == NULL) {
  		return NOT_OK_RESPONSE;
  	}
  
  	od_router_foreach(router, od_console_show_errors_per_route_cb, argv);
  
  	od_retcode_t rc = kiwi_be_write_complete(stream, "SHOW", 5);
  	return rc;
  }
  
  static inline int od_console_show_version(machine_msg_t *stream)
  {
  	assert(stream);
  
  	if (kiwi_be_write_row_descriptionf(stream, "s", "version") == NULL) {
  		return NOT_OK_RESPONSE;
  	}
  
  	int offset;
  	if (kiwi_be_write_data_row(stream, &offset) == NULL) {
  		return NOT_OK_RESPONSE;
  	}
  
  	char data[128];
  	int data_len;
  	/* current version and build */
  	data_len =
  		od_snprintf(data, sizeof(data), "%s-%s-%s", OD_VERSION_NUMBER,
  			    OD_VERSION_GIT, OD_VERSION_BUILD);
  
  	int rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc != OK_RESPONSE)
  		return rc;
  
  	rc = kiwi_be_write_complete(stream, "SHOW", 5);
  	return rc;
  }
  
  static inline od_retcode_t
  od_console_show_quantiles(machine_msg_t *stream, int offset,
  			  const int quantiles_count, const double *quantiles,
  			  td_histogram_t *transactions_hgram,
  			  td_histogram_t *queries_hgram)
  {
  	char data[64];
  	int data_len;
  	int rc = OK_RESPONSE;
  	for (int i = 0; i < quantiles_count; i++) {
  		double q = quantiles[i];
  		/* query quantile */
  		double query_quantile = td_value_at(queries_hgram, q);
  		double transaction_quantile =
  			td_value_at(transactions_hgram, q);
  		if (isnan(query_quantile)) {
  			query_quantile = 0;
  		}
  		if (isnan(transaction_quantile)) {
  			transaction_quantile = 0;
  		}
  		data_len = od_snprintf(data, sizeof(data), "%" PRIu64,
  				       (uint64_t)query_quantile);
  		rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  		if (rc == NOT_OK_RESPONSE)
  			return rc;
  		/* transaction quantile */
  		data_len = od_snprintf(data, sizeof(data), "%" PRIu64,
  				       (uint64_t)transaction_quantile);
  		rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  		if (rc == NOT_OK_RESPONSE)
  			return rc;
  	}
  	return rc;
  }
  
  static inline int od_console_show_pools_add_cb(od_route_t *route, void **argv)
  {
  	int offset;
  	machine_msg_t *stream = argv[0];
  	bool *extended = argv[1];
  	double *quantiles = argv[2];
  	int *quantiles_count = argv[3];
  	td_histogram_t *common_transactions_hgram = argv[4];
  	td_histogram_t *common_queries_hgram = argv[5];
  
  	machine_msg_t *msg;
  	td_histogram_t *transactions_hgram = NULL;
  	td_histogram_t *queries_hgram = NULL;
  	td_histogram_t *freeze_hgram = NULL;
  	msg = kiwi_be_write_data_row(stream, &offset);
  	if (msg == NULL)
  		return NOT_OK_RESPONSE;
  
  	od_route_lock(route);
  	int rc;
  	rc = kiwi_be_write_data_row_add(stream, offset, route->id.database,
  					route->id.database_len - 1);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  	rc = kiwi_be_write_data_row_add(stream, offset, route->id.user,
  					route->id.user_len - 1);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  	char data[64];
  	int data_len;
  
  	/* cl_active */
  	data_len = od_snprintf(data, sizeof(data), "%d",
  			       route->client_pool.count_active);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  	/* cl_waiting */
  	data_len = od_snprintf(data, sizeof(data), "%d",
  			       route->client_pool.count_pending);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  	/* sv_active */
  	data_len = od_snprintf(data, sizeof(data), "%d",
  			       route->server_pool.count_active);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  	/* sv_idle */
  	data_len = od_snprintf(data, sizeof(data), "%d",
  			       route->server_pool.count_idle);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  	/* sv_used */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, 0UL);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  	/* sv_tested */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, 0UL);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  	/* sv_login */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, 0UL);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  	/* maxwait */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, 0UL);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  	/* maxwait_us */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, 0UL);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  
  	/* pool_mode */
  	rc = -1;
  	if (route->rule->pool == OD_RULE_POOL_SESSION)
  		rc = kiwi_be_write_data_row_add(stream, offset, "session", 7);
  	if (route->rule->pool == OD_RULE_POOL_TRANSACTION)
  		rc = kiwi_be_write_data_row_add(stream, offset, "transaction",
  						11);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  
  	if (*extended) {
  		/* bytes recived */
  		data_len = od_snprintf(data, sizeof(data), "%" PRIu64,
  				       route->stats.recv_client);
  		rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  		if (rc == NOT_OK_RESPONSE)
  			goto error;
  		/* bytes sent */
  		data_len = od_snprintf(data, sizeof(data), "%" PRIu64,
  				       route->stats.recv_server);
  		rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  		if (rc == NOT_OK_RESPONSE)
  			goto error;
  
  		/* tcp conn rate */
  		data_len = od_snprintf(data, sizeof(data), "%" PRIu64,
  				       route->tcp_connections);
  		rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  		if (rc == NOT_OK_RESPONSE)
  			goto error;
  
  		transactions_hgram = td_new(QUANTILES_COMPRESSION);
  		queries_hgram = td_new(QUANTILES_COMPRESSION);
  		freeze_hgram = td_new(QUANTILES_COMPRESSION);
  		if (route->stats.enable_quantiles) {
  			for (size_t i = 0; i < QUANTILES_WINDOW; ++i) {
  				td_copy(freeze_hgram,
  					route->stats.transaction_hgram[i]);
  				td_merge(transactions_hgram, freeze_hgram);
  				td_copy(freeze_hgram,
  					route->stats.query_hgram[i]);
  				td_merge(queries_hgram, freeze_hgram);
  			}
  			td_merge(common_transactions_hgram, transactions_hgram);
  			td_merge(common_queries_hgram, queries_hgram);
  		}
  		rc = od_console_show_quantiles(stream, offset, *quantiles_count,
  					       quantiles, transactions_hgram,
  					       queries_hgram);
  		if (rc == NOT_OK_RESPONSE) {
  			goto error;
  		}
  	}
  	td_safe_free(transactions_hgram);
  	td_safe_free(queries_hgram);
  	td_safe_free(freeze_hgram);
  	od_route_unlock(route);
  	return 0;
  error:
  	td_safe_free(transactions_hgram);
  	td_safe_free(queries_hgram);
  	td_safe_free(freeze_hgram);
  	od_route_unlock(route);
  	return NOT_OK_RESPONSE;
  }
  
  static inline int od_console_show_databases_add_cb(od_route_t *route,
  						   void **argv)
  {
  	int offset;
  	machine_msg_t *stream = argv[0];
  	machine_msg_t *msg;
  	msg = kiwi_be_write_data_row(stream, &offset);
  	if (msg == NULL)
  		return NOT_OK_RESPONSE;
  
  	od_route_lock(route);
  	int rc;
  	rc = kiwi_be_write_data_row_add(stream, offset, route->id.database,
  					route->id.database_len - 1);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  	od_rule_t *rule = route->rule;
  	od_rule_storage_t *storage = rule->storage;
  
  	char *host = storage->host;
  	if (!host) {
  		host = "";
  	}
  
  	rc = kiwi_be_write_data_row_add(stream, offset, host, strlen(host));
  	if (rc == NOT_OK_RESPONSE) {
  		goto error;
  	}
  
  	char data[64];
  	int data_len;
  
  	/* port */
  	data_len = od_snprintf(data, sizeof(data), "%d", storage->port);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  
  	/* database */
  	rc = kiwi_be_write_data_row_add(stream, offset, rule->db_name,
  					rule->db_name_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  
  	/* force_user */
  	rc = kiwi_be_write_data_row_add(stream, offset, "", 0);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  
  	/* pool_size */
  	data_len = od_snprintf(data, sizeof(data), "%d", rule->pool_size);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  
  	/* reserve_pool */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, 0UL);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  
  	/* pool_mode */
  	rc = -1;
  	if (rule->pool == OD_RULE_POOL_SESSION)
  		rc = kiwi_be_write_data_row_add(stream, offset, "session", 7);
  	if (rule->pool == OD_RULE_POOL_TRANSACTION)
  		rc = kiwi_be_write_data_row_add(stream, offset, "transaction",
  						11);
  
  	if (rc == NOT_OK_RESPONSE) {
  		goto error;
  	}
  
  	/* max_connections */
  	data_len = od_snprintf(data, sizeof(data), "%d", rule->client_max);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  
  	/* current_connections */
  	data_len = od_snprintf(data, sizeof(data), "%d",
  			       route->client_pool.count_active +
  				       route->client_pool.count_pending +
  				       route->client_pool.count_queue);
  
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  
  	/* paused */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, 0UL);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  
  	/* disabled */
  	data_len = od_snprintf(data, sizeof(data), "%" PRIu64, 0UL);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  
  	od_route_unlock(route);
  	return 0;
  error:
  	od_route_unlock(route);
  	return NOT_OK_RESPONSE;
  }
  
  static inline int od_console_show_databases(od_client_t *client,
  					    machine_msg_t *stream)
  {
  	assert(stream);
  	od_router_t *router = client->global->router;
  
  	machine_msg_t *msg;
  	msg = kiwi_be_write_row_descriptionf(
  		stream, "sslssllsllll", "name", "host", "port", "database",
  		"force_user", "pool_size", "reserve_pool", "pool_mode",
  		"max_connections", "current_connections", "paused", "disabled");
  	if (msg == NULL)
  		return NOT_OK_RESPONSE;
  
  	void *argv[] = { stream };
  	int rc;
  	rc = od_router_foreach(router, od_console_show_databases_add_cb, argv);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  
  	return kiwi_be_write_complete(stream, "SHOW", 5);
  }
  
  static inline int od_console_show_pools(od_client_t *client,
  					machine_msg_t *stream, bool extended)
  {
  	assert(stream);
  	int rc;
  	od_router_t *router = client->global->router;
  	od_route_t *route = client->route;
  	double *quantiles = route->rule->quantiles;
  	int quantiles_count = route->rule->quantiles_count;
  
  	machine_msg_t *msg;
  	msg = kiwi_be_write_row_descriptionf(stream, "ssllllllllls", "database",
  					     "user", "cl_active", "cl_waiting",
  					     "sv_active", "sv_idle", "sv_used",
  					     "sv_tested", "sv_login", "maxwait",
  					     "maxwait_us", "pool_mode");
  	if (msg == NULL)
  		return NOT_OK_RESPONSE;
  
  	if (extended) {
  		char *bytes_rcv = "bytes_recieved";
  		rc = kiwi_be_write_row_description_add(msg, 0, bytes_rcv,
  						       strlen(bytes_rcv), 0, 0,
  						       23 /* INT4OID */, 4, 0,
  						       0);
  		if (rc == NOT_OK_RESPONSE)
  			return NOT_OK_RESPONSE;
  		char *bytes_sent = "bytes_sent";
  		rc = kiwi_be_write_row_description_add(msg, 0, bytes_sent,
  						       strlen(bytes_sent), 0, 0,
  						       23 /* INT4OID */, 4, 0,
  						       0);
  		if (rc == NOT_OK_RESPONSE)
  			return NOT_OK_RESPONSE;
  
  		char *tcp_conn_rate = "tcp_conn_count";
  		rc = kiwi_be_write_row_description_add(msg, 0, tcp_conn_rate,
  						       strlen(tcp_conn_rate), 0,
  						       0, 23 /* INT4OID */, 4,
  						       0, 0);
  		if (rc == NOT_OK_RESPONSE)
  			return NOT_OK_RESPONSE;
  
  		for (int i = 0; i < quantiles_count; i++) {
  			char caption[KIWI_MAX_VAR_SIZE];
  			int caption_len;
  			caption_len = od_snprintf(caption, sizeof(caption),
  						  "query_%.6g", quantiles[i]);
  			rc = kiwi_be_write_row_description_add(
  				msg, 0, caption, caption_len, 0, 0,
  				23 /* INT4OID */, 4, 0, 0);
  			if (rc == NOT_OK_RESPONSE)
  				return NOT_OK_RESPONSE;
  			caption_len =
  				od_snprintf(caption, sizeof(caption),
  					    "transaction_%.6g", quantiles[i]);
  			rc = kiwi_be_write_row_description_add(
  				msg, 0, caption, caption_len, 0, 0,
  				23 /* INT4OID */, 4, 0, 0);
  			if (rc == NOT_OK_RESPONSE)
  				return NOT_OK_RESPONSE;
  		}
  	}
  
  	td_histogram_t *transactions_hgram = NULL;
  	td_histogram_t *queries_hgram = NULL;
  	if (extended) {
  		transactions_hgram = td_new(QUANTILES_COMPRESSION);
  		queries_hgram = td_new(QUANTILES_COMPRESSION);
  	}
  	void *argv[] = { stream,	   &extended,	       quantiles,
  			 &quantiles_count, transactions_hgram, queries_hgram };
  	rc = od_router_foreach(router, od_console_show_pools_add_cb, argv);
  	if (rc == NOT_OK_RESPONSE)
  		goto error;
  	if (extended) {
  		int offset;
  		msg = kiwi_be_write_data_row(stream, &offset);
  		if (msg == NULL)
  			goto error;
  		char *aggregated_name = "aggregated";
  		rc = kiwi_be_write_data_row_add(stream, offset, aggregated_name,
  						strlen(aggregated_name));
  		if (rc == NOT_OK_RESPONSE) {
  			goto error;
  		}
  		rc = kiwi_be_write_data_row_add(stream, offset, aggregated_name,
  						strlen(aggregated_name));
  		if (rc == NOT_OK_RESPONSE) {
  			goto error;
  		}
  		const size_t rest_columns_count = 13;
  		for (size_t i = 0; i < rest_columns_count; ++i) {
  			rc = kiwi_be_write_data_row_add(stream, offset, NULL,
  							NULL_MSG_LEN);
  			if (rc == NOT_OK_RESPONSE) {
  				goto error;
  			}
  		}
  		rc = od_console_show_quantiles(stream, offset, quantiles_count,
  					       quantiles, transactions_hgram,
  					       queries_hgram);
  		if (rc == NOT_OK_RESPONSE) {
  			goto error;
  		}
  	}
  	td_safe_free(transactions_hgram);
  	td_safe_free(queries_hgram);
  	return kiwi_be_write_complete(stream, "SHOW", 5);
  error:
  	td_safe_free(transactions_hgram);
  	td_safe_free(queries_hgram);
  	return NOT_OK_RESPONSE;
  }
  
  static inline int od_console_show_servers_server_cb(od_server_t *server,
  						    void **argv)
  {
  	od_route_t *route = server->route;
  
  	int offset;
  	machine_msg_t *stream = argv[0];
  	machine_msg_t *msg;
  	msg = kiwi_be_write_data_row(stream, &offset);
  	if (msg == NULL)
  		return NOT_OK_RESPONSE;
  	/* type */
  	char data[64];
  	size_t data_len;
  	data_len = od_snprintf(data, sizeof(data), "S");
  	int rc;
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* user */
  	rc = kiwi_be_write_data_row_add(stream, offset, route->id.user,
  					route->id.user_len - 1);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* database */
  	rc = kiwi_be_write_data_row_add(stream, offset, route->id.database,
  					route->id.database_len - 1);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* state */
  	char *state = "";
  	if (server->state == OD_SERVER_IDLE)
  		state = "idle";
  	else if (server->state == OD_SERVER_ACTIVE)
  		state = "active";
  	rc = kiwi_be_write_data_row_add(stream, offset, state, strlen(state));
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* addr */
  	od_getpeername(server->io.io, data, sizeof(data), 1, 0);
  	data_len = strlen(data);
  	rc = kiwi_be_write_data_row_add(msg, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* port */
  	od_getpeername(server->io.io, data, sizeof(data), 0, 1);
  	data_len = strlen(data);
  	rc = kiwi_be_write_data_row_add(msg, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* local_addr */
  	od_getsockname(server->io.io, data, sizeof(data), 1, 0);
  	data_len = strlen(data);
  	rc = kiwi_be_write_data_row_add(msg, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* local_port */
  	od_getsockname(server->io.io, data, sizeof(data), 0, 1);
  	data_len = strlen(data);
  	rc = kiwi_be_write_data_row_add(msg, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* connect_time */
  	rc = kiwi_be_write_data_row_add(msg, offset, NULL, -1);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* request_time */
  	rc = kiwi_be_write_data_row_add(msg, offset, NULL, -1);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* wait */
  	data_len = od_snprintf(data, sizeof(data), "0");
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* wait_us */
  	data_len = od_snprintf(data, sizeof(data), "0");
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* ptr */
  	data_len =
  		od_snprintf(data, sizeof(data), "%s%.*s", server->id.id_prefix,
  			    (signed)sizeof(server->id.id), server->id.id);
  	rc = kiwi_be_write_data_row_add(msg, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* link */
  	data_len = od_snprintf(data, sizeof(data), "%s", "");
  	rc = kiwi_be_write_data_row_add(msg, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* remote_pid */
  	data_len = od_snprintf(data, sizeof(data), "0");
  	rc = kiwi_be_write_data_row_add(msg, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* tls */
  	data_len = od_snprintf(data, sizeof(data), "%s", "");
  	rc = kiwi_be_write_data_row_add(msg, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	return 0;
  }
  
  static inline int od_console_show_servers_cb(od_route_t *route, void **argv)
  {
  	od_route_lock(route);
  
  	od_server_pool_foreach(&route->server_pool, OD_SERVER_ACTIVE,
  			       od_console_show_servers_server_cb, argv);
  
  	od_server_pool_foreach(&route->server_pool, OD_SERVER_IDLE,
  			       od_console_show_servers_server_cb, argv);
  
  	od_route_unlock(route);
  	return 0;
  }
  
  static inline int od_console_show_servers(od_client_t *client,
  					  machine_msg_t *stream)
  {
  	assert(stream);
  	od_router_t *router = client->global->router;
  
  	machine_msg_t *msg;
  	msg = kiwi_be_write_row_descriptionf(
  		stream, "sssssdsdssddssds", "type", "user", "database", "state",
  		"addr", "port", "local_addr", "local_port", "connect_time",
  		"request_time", "wait", "wait_us", "ptr", "link", "remote_pid",
  		"tls");
  	if (msg == NULL)
  		return NOT_OK_RESPONSE;
  
  	void *argv[] = { stream };
  	od_router_foreach(router, od_console_show_servers_cb, argv);
  
  	return kiwi_be_write_complete(stream, "SHOW", 5);
  }
  
  static inline int od_console_show_clients_callback(od_client_t *client,
  						   void **argv)
  {
  	int offset;
  	machine_msg_t *stream = argv[0];
  	machine_msg_t *msg;
  	msg = kiwi_be_write_data_row(stream, &offset);
  	if (msg == NULL)
  		return NOT_OK_RESPONSE;
  	char data[64];
  	size_t data_len;
  	/* type */
  	data_len = od_snprintf(data, sizeof(data), "C");
  	int rc;
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* user */
  	rc = kiwi_be_write_data_row_add(stream, offset,
  					client->startup.user.value,
  					client->startup.user.value_len - 1);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* database */
  	rc = kiwi_be_write_data_row_add(stream, offset,
  					client->startup.database.value,
  					client->startup.database.value_len - 1);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* state */
  	char *state = "";
  	if (client->state == OD_CLIENT_ACTIVE)
  		state = "active";
  	else if (client->state == OD_CLIENT_PENDING)
  		state = "pending";
  	else if (client->state == OD_CLIENT_QUEUE)
  		state = "queue";
  	rc = kiwi_be_write_data_row_add(stream, offset, state, strlen(state));
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* addr */
  	od_getpeername(client->io.io, data, sizeof(data), 1, 0);
  	data_len = strlen(data);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* port */
  	od_getpeername(client->io.io, data, sizeof(data), 0, 1);
  	data_len = strlen(data);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* local_addr */
  	od_getsockname(client->io.io, data, sizeof(data), 1, 0);
  	data_len = strlen(data);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* local_port */
  	od_getsockname(client->io.io, data, sizeof(data), 0, 1);
  	data_len = strlen(data);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* connect_time */
  	rc = kiwi_be_write_data_row_add(stream, offset, NULL, -1);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* request_time */
  	rc = kiwi_be_write_data_row_add(stream, offset, NULL, -1);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* wait */
  	data_len = od_snprintf(data, sizeof(data), "0");
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* wait_us */
  	data_len = od_snprintf(data, sizeof(data), "0");
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* ptr */
  	data_len =
  		od_snprintf(data, sizeof(data), "%s%.*s", client->id.id_prefix,
  			    (signed)sizeof(client->id.id), client->id.id);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* link */
  	data_len = od_snprintf(data, sizeof(data), "%s", "");
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* remote_pid */
  	data_len = od_snprintf(data, sizeof(data), "0");
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* tls */
  	data_len = od_snprintf(data, sizeof(data), "%s", "");
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	return 0;
  }
  
  static inline int od_console_show_clients_cb(od_route_t *route, void **argv)
  {
  	od_route_lock(route);
  
  	od_client_pool_foreach(&route->client_pool, OD_CLIENT_ACTIVE,
  			       od_console_show_clients_callback, argv);
  
  	od_client_pool_foreach(&route->client_pool, OD_CLIENT_PENDING,
  			       od_console_show_clients_callback, argv);
  
  	od_client_pool_foreach(&route->client_pool, OD_CLIENT_QUEUE,
  			       od_console_show_clients_callback, argv);
  
  	od_route_unlock(route);
  	return 0;
  }
  
  static inline int od_console_show_clients(od_client_t *client,
  					  machine_msg_t *stream)
  {
  	assert(stream);
  	od_router_t *router = client->global->router;
  
  	machine_msg_t *msg;
  	msg = kiwi_be_write_row_descriptionf(
  		stream, "sssssdsdssddssds", "type", "user", "database", "state",
  		"addr", "port", "local_addr", "local_port", "connect_time",
  		"request_time", "wait", "wait_us", "ptr", "link", "remote_pid",
  		"tls");
  	if (msg == NULL)
  		return NOT_OK_RESPONSE;
  
  	void *argv[] = { stream };
  	od_router_foreach(router, od_console_show_clients_cb, argv);
  
  	return kiwi_be_write_complete(stream, "SHOW", 5);
  }
  
  static inline int od_console_show_lists_add(machine_msg_t *stream, char *list,
  					    int items)
  {
  	int offset;
  	machine_msg_t *msg;
  	msg = kiwi_be_write_data_row(stream, &offset);
  	if (msg == NULL)
  		return NOT_OK_RESPONSE;
  	/* list */
  	int rc;
  	rc = kiwi_be_write_data_row_add(stream, offset, list, strlen(list));
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* items */
  	char data[64];
  	int data_len;
  	data_len = od_snprintf(data, sizeof(data), "%d", items);
  	rc = kiwi_be_write_data_row_add(stream, offset, data, data_len);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	return 0;
  }
  
  static inline int od_console_show_lists_cb(od_route_t *route, void **argv)
  {
  	od_route_lock(route);
  
  	int *used_servers = argv[0];
  	int *free_servers = argv[1];
  	(*used_servers) += route->server_pool.count_active;
  	(*free_servers) += route->server_pool.count_idle;
  
  	od_route_unlock(route);
  	return 0;
  }
  
  static inline int od_console_show_lists(od_client_t *client,
  					machine_msg_t *stream)
  {
  	assert(stream);
  	od_router_t *router = client->global->router;
  
  	/* Gather router information.
  
  	   router_used_servers can be inconsistent here, since it depends on
  	   separate route locks.
  	*/
  	od_router_lock(router);
  
  	int router_used_servers = 0;
  	int router_free_servers = 0;
  	int router_pools = router->route_pool.count;
  	int router_clients = od_atomic_u32_of(&router->clients);
  
  	void *argv[] = { &router_used_servers, &router_free_servers };
  	od_route_pool_foreach(&router->route_pool, od_console_show_lists_cb,
  			      argv);
  
  	od_router_unlock(router);
  
  	machine_msg_t *msg;
  	msg = kiwi_be_write_row_descriptionf(stream, "sd", "list", "items");
  	if (msg == NULL)
  		return NOT_OK_RESPONSE;
  	int rc;
  	/* databases */
  	rc = od_console_show_lists_add(stream, "databases", 0);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* users */
  	rc = od_console_show_lists_add(stream, "users", 0);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* pools */
  	rc = od_console_show_lists_add(stream, "pools", router_pools);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* free_clients */
  	rc = od_console_show_lists_add(stream, "free_clients", 0);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* used_clients */
  	rc = od_console_show_lists_add(stream, "used_clients", router_clients);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* login_clients */
  	rc = od_console_show_lists_add(stream, "login_clients", 0);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* free_servers */
  	rc = od_console_show_lists_add(stream, "free_servers",
  				       router_free_servers);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* used_servers */
  	rc = od_console_show_lists_add(stream, "used_servers",
  				       router_used_servers);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* dns_names */
  	rc = od_console_show_lists_add(stream, "dns_names", 0);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* dns_zones */
  	rc = od_console_show_lists_add(stream, "dns_zones", 0);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* dns_queries */
  	rc = od_console_show_lists_add(stream, "dns_queries", 0);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	/* dns_pending */
  	rc = od_console_show_lists_add(stream, "dns_pending", 0);
  	if (rc == NOT_OK_RESPONSE)
  		return NOT_OK_RESPONSE;
  	return kiwi_be_write_complete(stream, "SHOW", 5);
  }
  
  static inline int od_console_show(od_client_t *client, machine_msg_t *stream,
  				  od_parser_t *parser)
  {
  	assert(stream);
  	od_token_t token;
  	int rc;
  	rc = od_parser_next(parser, &token);
  	switch (rc) {
  	case OD_PARSER_KEYWORD:
  		break;
  	case OD_PARSER_EOF:
  	default:
  		return NOT_OK_RESPONSE;
  	}
  	od_keyword_t *keyword;
  	keyword = od_keyword_match(od_console_keywords, &token);
  	if (keyword == NULL)
  		return NOT_OK_RESPONSE;
  	switch (keyword->id) {
  	case OD_LSTATS:
  		return od_console_show_stats(client, stream);
  	case OD_LPOOLS:
  		return od_console_show_pools(client, stream, false);
  	case OD_LPOOLS_EXTENDED:
  		return od_console_show_pools(client, stream, true);
  	case OD_LDATABASES:
  		return od_console_show_databases(client, stream);
  	case OD_LSERVERS:
  		return od_console_show_servers(client, stream);
  	case OD_LCLIENTS:
  		return od_console_show_clients(client, stream);
  	case OD_LLISTS:
  		return od_console_show_lists(client, stream);
  	case OD_LERRORS:
  		return od_console_show_errors(client, stream);
  	case OD_LERRORS_PER_ROUTE:
  		return od_console_show_errors_per_route(client, stream);
  	case OD_LVERSION:
  		return od_console_show_version(stream);
  	}
  	return NOT_OK_RESPONSE;
  }
  
  static inline int od_console_kill_client(od_client_t *client,
  					 machine_msg_t *stream,
  					 od_parser_t *parser)
  {
  	(void)stream;
  	od_token_t token;
  	int rc;
  	rc = od_parser_next(parser, &token);
  	if (rc != OD_PARSER_KEYWORD)
  		return NOT_OK_RESPONSE;
  	od_id_t id;
  	if (token.value.string.size != (sizeof(id.id) + 1))
  		return NOT_OK_RESPONSE;
  	memcpy(id.id, token.value.string.pointer + 1, sizeof(id.id));
  
  	od_router_kill(client->global->router, &id);
  	return 0;
  }
  
  static inline int od_console_reload(od_client_t *client, machine_msg_t *stream)
  {
  	od_instance_t *instance = client->global->instance;
  
  	od_log(&instance->logger, "console", NULL, NULL,
  	       "RELOAD command received");
  	od_system_config_reload(client->global->system);
  	return kiwi_be_write_complete(stream, "RELOAD", 7);
  }
  
  static inline int od_console_set(od_client_t *client, machine_msg_t *stream)
  {
  	(void)client;
  	/* reply success */
  	return kiwi_be_write_complete(stream, "SET", 4);
  }
  
  static inline int od_console_add_module(od_client_t *client,
  					machine_msg_t *stream,
  					od_parser_t *parser)
  {
  	assert(stream);
  	od_token_t token;
  	int rc;
  	rc = od_parser_next(parser, &token);
  	od_instance_t *instance = client->global->instance;
  
  	switch (rc) {
  	case OD_PARSER_STRING: {
  		char module_path[MAX_MODULE_PATH_LEN];
  		od_token_to_string_dest(&token, module_path);
  
  		od_log(&instance->logger, "od module dynamic load", NULL, NULL,
  		       "loading module with path %s", module_path);
  		int retcode = od_target_module_add(
  			&instance->logger,
  			((od_extention_t *)client->global->extentions)->modules,
  			module_path);
  		if (retcode == 0) {
  			od_frontend_infof(client, stream,
  					  "module was successfully loaded!");
  		} else {
  			od_frontend_errorf(
  				client, stream, KIWI_SYSTEM_ERROR,
  				"module was NOT successfully loaded! Check logs for details");
  		}
  		return retcode;
  	}
  	case OD_PARSER_EOF:
  	default:
  		return NOT_OK_RESPONSE;
  	}
  }
  
  static inline int od_console_unload_module(od_client_t *client,
  					   machine_msg_t *stream,
  					   od_parser_t *parser)
  {
  	assert(stream);
  	od_token_t token;
  	int rc;
  	rc = od_parser_next(parser, &token);
  	od_instance_t *instance = client->global->instance;
  
  	switch (rc) {
  	case OD_PARSER_STRING: {
  		char module_path[MAX_MODULE_PATH_LEN];
  		od_token_to_string_dest(&token, module_path);
  
  		od_log(&instance->logger, "od module dynamic unload", NULL,
  		       NULL, "unloading module with path %s", module_path);
  		int retcode = od_target_module_unload(
  			&instance->logger,
  			((od_extention_t *)client->global->extentions)->modules,
  			module_path);
  		if (retcode == 0) {
  			od_frontend_infof(client, stream,
  					  "module was successfully unloaded!");
  		} else {
  			od_frontend_errorf(client, stream, KIWI_SYSTEM_ERROR,
  					   "module was NOT successfully "
  					   "unloaded! Check logs for details");
  		}
  		return retcode;
  	}
  	case OD_PARSER_EOF:
  	default:
  		return NOT_OK_RESPONSE;
  	}
  }
  
  static inline int od_console_create(od_client_t *client, machine_msg_t *stream,
  				    od_parser_t *parser)
  {
  	assert(stream);
  	od_token_t token;
  	int rc;
  	rc = od_parser_next(parser, &token);
  	switch (rc) {
  	case OD_PARSER_KEYWORD:
  		break;
  	case OD_PARSER_EOF:
  	default:
  		return NOT_OK_RESPONSE;
  	}
  	od_keyword_t *keyword;
  	keyword = od_keyword_match(od_console_keywords, &token);
  	if (keyword == NULL)
  		return NOT_OK_RESPONSE;
  
  	switch (keyword->id) {
  	case OD_LMODULE:
  		return od_console_add_module(client, stream, parser);
  	}
  
  	return NOT_OK_RESPONSE;
  }
  
  static inline int od_console_drop(od_client_t *client, machine_msg_t *stream,
  				  od_parser_t *parser)
  {
  	assert(stream);
  	od_token_t token;
  	int rc;
  	rc = od_parser_next(parser, &token);
  	switch (rc) {
  	case OD_PARSER_KEYWORD:
  		break;
  	case OD_PARSER_EOF:
  	default:
  		return NOT_OK_RESPONSE;
  	}
  	od_keyword_t *keyword;
  	keyword = od_keyword_match(od_console_keywords, &token);
  	if (keyword == NULL)
  		return NOT_OK_RESPONSE;
  
  	switch (keyword->id) {
  	case OD_LMODULE:
  		return od_console_unload_module(client, stream, parser);
  	}
  
  	return NOT_OK_RESPONSE;
  }
  
  int od_console_query(od_client_t *client, machine_msg_t *stream,
  		     char *query_data, uint32_t query_data_size)
  {
  	od_instance_t *instance = client->global->instance;
  
  	uint32_t query_len;
  	char *query;
  	machine_msg_t *msg;
  	int rc;
  	rc = kiwi_be_read_query(query_data, query_data_size, &query,
  				&query_len);
  	if (rc == NOT_OK_RESPONSE) {
  		od_error(&instance->logger, "console", client, NULL,
  			 "bad console command");
  		msg = od_frontend_errorf(client, stream, KIWI_SYNTAX_ERROR,
  					 "bad console command");
  		if (msg == NULL)
  			return NOT_OK_RESPONSE;
  
  		return 0;
  	}
  
  	if (instance->config.log_query)
  		od_debug(&instance->logger, "console", client, NULL, "%.*s",
  			 query_len, query);
  
  	od_parser_t parser;
  	od_parser_init(&parser, query, query_len);
  
  	od_token_t token;
  	rc = od_parser_next(&parser, &token);
  	switch (rc) {
  	case OD_PARSER_KEYWORD:
  		break;
  	case OD_PARSER_EOF:
  	default:
  		goto bad_query;
  	}
  	od_keyword_t *keyword;
  	keyword = od_keyword_match(od_console_keywords, &token);
  	if (keyword == NULL)
  		goto bad_query;
  	switch (keyword->id) {
  	case OD_LSHOW:
  		rc = od_console_show(client, stream, &parser);
  		if (rc == NOT_OK_RESPONSE)
  			goto bad_query;
  		break;
  	case OD_LKILL_CLIENT:
  		rc = od_console_kill_client(client, stream, &parser);
  		if (rc == NOT_OK_RESPONSE)
  			goto bad_query;
  		break;
  	case OD_LRELOAD:
  		rc = od_console_reload(client, stream);
  		if (rc == NOT_OK_RESPONSE)
  			goto bad_query;
  		break;
  	case OD_LSET:
  		rc = od_console_set(client, stream);
  		if (rc == NOT_OK_RESPONSE)
  			goto bad_query;
  		break;
  	case OD_LCREATE:
  		rc = od_console_create(client, stream, &parser);
  		if (rc == NOT_OK_RESPONSE) {
  			goto bad_query;
  		}
  		break;
  	case OD_LDROP:
  		rc = od_console_drop(client, stream, &parser);
  		if (rc == NOT_OK_RESPONSE) {
  			goto bad_query;
  		}
  		break;
  	default:
  		goto bad_query;
  	}
  
  	return 0;
  
  bad_query:
  	od_error(&instance->logger, "console", client, NULL,
  		 "console command error: %.*s", query_len, query);
  
  	msg = od_frontend_errorf(client, stream, KIWI_SYNTAX_ERROR,
  				 "console command error: %.*s", query_len,
  				 query);
  	if (msg == NULL)
  		return NOT_OK_RESPONSE;
  
  	return 0;
  }
  