static void
gst_dtls_dec_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec)
{
  GstDtlsDec *self = GST_DTLS_DEC (object);

  switch (prop_id) {
    case PROP_CONNECTION_ID:
      g_free (self->connection_id);
      self->connection_id = g_value_dup_string (value);
      g_return_if_fail (self->agent);
      create_connection (self, self->connection_id);
      break;
    case PROP_PEM:
      if (self->agent) {
        g_object_unref (self->agent);
      }
      self->agent = get_agent_by_pem (g_value_get_string (value));
      if (self->connection_id) {
        create_connection (self, self->connection_id);
      }
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (self, prop_id, pspec);
  }
}

void create_client(unsigned int part_size, char *message)
{
	int write_length;

	size_t text_length;
	char text[MAX_MESSAGE_SIZE];

	if (message) {
		text_length = strlen(message);
		
		if (!simulate_sending) {
			create_connection();
			LOG(DEBUG, "Connection created");
		}

		strncpy(text, message, MAX_MESSAGE_SIZE);

		part_size = prepare_connection(text_length, part_size);
		if (part_size == 0)
			die("preparing connection");

		LOG(DEBUG, "{ Message from command line to send: '%s' (%u)", text, text_length);
		write_length = send_message(/*connection, */text, text_length, part_size);
		LOG(DEBUG, "} Message written; packet size: %i", write_length);

		if (!simulate_sending)
			close_connection();

		LOG(INFO, "Nothing more to do. Exiting.");
	} else {

		LOG(INFO, "Type message and press enter");

		for (;;) {
			printf(CLIENT_PROMPT);

			text_length = read_line(text, MAX_MESSAGE_SIZE);

			if (!simulate_sending) {
				create_connection();
				LOG(DEBUG, "Connection created");
			}

			part_size = prepare_connection(text_length, part_size);
			if (part_size == 0)
				die("preparing connection");

			LOG(DEBUG, "Message to send: '%s' (%u)", text, text_length);

			write_length = send_message(/*connection, */text, text_length, part_size);
			//sleep(1); // Prevent client from spamming

			if (!simulate_sending)
				close_connection();
		}

		LOG(INFO, "Server closed connection?");
	}
}

connection_pool::connection_pool(int _size,
                                 const std::string& _host,
                                 const int _port,
                                 const std::string& _username,
                                 const std::string& _zone,
                                 const int _refresh_time)
    : host_{_host}
    , port_{_port}
    , username_{_username}
    , zone_{_zone}
    , refresh_time_(_refresh_time)
    , conn_ctxs_(_size)
{
    if (_size < 1) {
        throw std::runtime_error{"invalid connection pool size"};
    }

    // Always initialize the first connection to guarantee that the
    // network plugin is loaded. This guarantees that asynchronous calls
    // to rcConnect do not cause a segfault.
    create_connection(0,
                      [] { throw std::runtime_error{"connect error"}; },
                      [] { throw std::runtime_error{"client login error"}; });

    // If the size of the pool is one, then return immediately.
    if (_size == 1) {
        return;
    }

    // Initialize the rest of the connection pool asynchronously.

    irods::thread_pool thread_pool{std::min<int>(_size, std::thread::hardware_concurrency())};

    std::atomic<bool> connect_error{};
    std::atomic<bool> login_error{};

    for (int i = 1; i < _size; ++i) {
        irods::thread_pool::post(thread_pool, [this, i, &connect_error, &login_error] {
            if (connect_error.load() || login_error.load()) {
                return;
            }

            create_connection(i,
                              [&connect_error] { connect_error.store(true); },
                              [&login_error] { login_error.store(true); });
        });
    }

    thread_pool.join();

    if (connect_error.load()) {
        throw std::runtime_error{"connect error"};
    }

    if (login_error.load()) {
        throw std::runtime_error{"client login error"};
    }
}

/* Search and connect
 * Returns:
 *   -1 - critical error (exit persist mode)
 *   1  - non critical error
 *   0  - success
 */
static int do_connect(void)
{
	inquiry_info *ii;
	int reconnect = 0;
	int i, n, r = 0;

	do {
		if (reconnect)
			sleep(persist);
		reconnect = 1;

		if (cache.valid > 0) {
			/* Use cached bdaddr */
			r = create_connection(cache.dst, &cache.bdaddr);
			if (r < 0) {
				terminate = 1;
				break;
			}
			continue;
		}

		syslog(LOG_INFO, "Inquiring");

		/* FIXME: Should we use non general LAP here ? */

		ii = NULL;
		n  = hci_inquiry(src_dev, search_duration, 10, NULL, &ii, 0);
		if (n < 0) {
			syslog(LOG_ERR, "Inquiry failed. %s(%d)", strerror(errno), errno);
			continue;
		}

		for (i = 0; i < n; i++) {
			char dst[40];
			ba2str(&ii[i].bdaddr, dst);

			if (use_sdp) {
				syslog(LOG_INFO, "Searching for %s on %s", 
						bnep_svc2str(service), dst);

				if (bnep_sdp_search(&src_addr, &ii[i].bdaddr, service) <= 0)
					continue;
			}

			r = create_connection(dst, &ii[i].bdaddr);
			if (r < 0) {
				terminate = 1;
				break;
			}
		}
		free(ii);
	} while (!terminate && persist);

	return r;
}

/*
** Connects to a data source.
*/
static int env_connect(lua_State *L)
{
  const char *sourcename;
  sqlite3 *conn;
  const char *errmsg;
  int res;
  getenvironment(L);  /* validate environment */

  sourcename = luaL_checkstring(L, 2);

  res = sqlite3_open(sourcename, &conn);
  if (res != SQLITE_OK)
    {
      errmsg = sqlite3_errmsg(conn);
      luasql_faildirect(L, errmsg);
      sqlite3_close(conn);
      return 2;
    }

  if (lua_isnumber(L, 3)) {
  	sqlite3_busy_timeout(conn, lua_tonumber(L,3)); // TODO: remove this
  }

  return create_connection(L, 1, conn);
}

TEST_F(positive_test_connection_to_levin_protocol_handler_calls, handler_processes_handle_read_as_notify)
{
  // Setup
  const int expected_command = 4673261;

  test_connection_ptr conn = create_connection();

  std::string in_data(256, 'e');

  epee::levin::bucket_head2 req_head;
  req_head.m_signature = LEVIN_SIGNATURE;
  req_head.m_cb = in_data.size();
  req_head.m_have_to_return_data = false;
  req_head.m_command = expected_command;
  req_head.m_flags = LEVIN_PACKET_REQUEST;
  req_head.m_protocol_version = LEVIN_PROTOCOL_VER_1;

  std::string buf(reinterpret_cast<const char*>(&req_head), sizeof(req_head));
  buf += in_data;

  // Test
  ASSERT_TRUE(conn->m_protocol_handler.handle_recv(buf.data(), buf.size()));

  // Check connection and levin_commands_handler states
  ASSERT_EQ(1, m_commands_handler.notify_counter());
  ASSERT_EQ(0, m_commands_handler.invoke_counter());
  ASSERT_EQ(expected_command, m_commands_handler.last_command());
  ASSERT_EQ(in_data, m_commands_handler.last_in_buf());
  ASSERT_LE(0, conn->send_counter());
  ASSERT_TRUE(conn->last_send_data().empty());
}

/* Search and connect
 * Returns:
 *   -1 - critical error (exit persist mode)
 *   1  - non critical error
 *   0  - success
 */
static int do_connect(bdaddr_t* src,bdaddr_t* dst)
{
  //	inquiry_info *ii;
	int reconnect = 0;
	//	int i, n;
	int r = 0;

	do {
		if (reconnect)
			sleep(persist);
		reconnect = 1;

		//TODO if (cache.valid > 0) {
			/* Use cached bdaddr */
			// r = create_connection(cache.dst, &cache.bdaddr);
                        r = create_connection(src,dst);
			if (r < 0) {
				terminate = 1;
				break;
			}
			continue;
		//TODO }

	} while (!terminate && persist);

	return r;
}

/* Handle client connection */
void handle_client(int client_sock, struct sockaddr_in client_addr)
{
    if ((remote_sock = create_connection()) < 0) {
        perror("Cannot connect to host");
        return;
    }

    init_clinet_ssl(remote_sock);//get ssl

    if (fork() == 0) { // a process forwarding data from client to remote socket

        forward_data_up(client_sock, remote_sock);

        exit(0);
    }

    if (fork() == 0) { // a process forwarding data from remote socket to client

        forward_data_down(remote_sock, client_sock);

        exit(0);
    }

    close(remote_sock);
    close(client_sock);

    SSL_free(ssl);
    SSL_CTX_free(ctx);
}

static void hci_link_control(uint16_t ocf, int plen, uint8_t *data)
{
	uint8_t status;

	const uint16_t ogf = OGF_LINK_CTL;

	switch (ocf) {
	case OCF_CREATE_CONN:
		command_status(ogf, ocf, 0x00);
		create_connection(data);
		break;

	case OCF_ACCEPT_CONN_REQ:
		command_status(ogf, ocf, 0x00);
		accept_connection(data);
		break;

	case OCF_DISCONNECT:
		command_status(ogf, ocf, 0x00);
		disconnect(data);
		break;

	default:
		status = 0x01;
		command_complete(ogf, ocf, 1, &status);
		break;
	}
}

/*
** Creates and returns a connection object
** Lua Input: source [, user [, pass]]
**   source: data source
**   user, pass: data source authentication information
** Lua Returns:
**   connection object if successfull
**   nil and error message otherwise.
*/
static int env_connect (lua_State *L) {
	env_data *env = (env_data *) getenvironment (L);
	const char *sourcename = luaL_checkstring (L, 2);
	const char *username = luaL_optstring (L, 3, NULL);
	const char *password = luaL_optstring (L, 4, NULL);
	SQLHDBC hdbc;
	SQLRETURN ret;

	/* tries to allocate connection handle */
	ret = SQLAllocHandle (hDBC, env->henv, &hdbc);
	if (error(ret))
		return luasql_faildirect (L, "connection allocation error.");

	/* tries to connect handle */
	ret = SQLConnect (hdbc, (char *) sourcename, SQL_NTS, 
		(char *) username, SQL_NTS, (char *) password, SQL_NTS);
	if (error(ret)) {
		ret = fail(L, hDBC, hdbc);
		SQLFreeHandle(hDBC, hdbc);
		return ret;
	}

	/* success, return connection object */
	return create_connection (L, 1, env, hdbc);
}

static void
service_added_cb (DMAPMdnsBrowser *browser,
                  DMAPMdnsBrowserService *service,
                  gpointer user_data)
{
	char answer, newline;

	if (NULL == service_name) {
		fprintf (stdout,
		    "service added %s:%s:%s:%d (%s)\n",
		     service->service_name,
		     service->name,
		     service->host,
		     service->port,
		     service->password_protected ? "protected" : "not protected");
		fprintf (stdout, "Stress test this service [Y|N]? ");
		fscanf (stdin, "%c%c", &answer, &newline);
	} else {
		answer = strcmp (service->service_name, service_name) ? 'N' : 'Y';
	}

	if (answer == 'Y') {
		create_connection (dmap_mdns_browser_get_service_type (browser),
				   service->name,
				   service->host,
				   service->port);
	}
}

static void server_loop(void) {
    int rc;

    chip_unipro_init();

    switch_set_local_dev_id(NULL, SWITCH_PORT_ID, LOCAL_DEV_ID);

    dbgprint("Wait for peer...\n");
    chip_reset_before_ready();

    rc = svc_wait_for_peer_ready();
    if (rc) {
        return;
    }

#if _SPECIAL_TEST == SPECIAL_GEAR_CHANGE_TEST
    switch_gear_change(GEAR_HS_G1,
                       TERMINATION_ON,
                       HS_MODE_A,
                       2,
                       POWERMODE_FAST);
#endif

    switch_if_dev_id_set(NULL, PEER_PORT_ID, PEER_DEV_ID);

    create_connection(&conn[0]);
    dbgprint("Control port connected\n");

    gb_control();
    gbboot_process();
#if _SPECIAL_TEST == SPECIAL_GBBOOT_SERVER_STANDBY
    if (stage_to_load == FFFF_ELEMENT_STAGE_2_FW)
        chip_enter_hibern8_server();
#endif
}

    void accept() override {
      auto session = std::make_shared<Session>(create_connection(*io_service, context));

      acceptor->async_accept(session->connection->socket->lowest_layer(), [this, session](const error_code &ec) {
        auto lock = session->connection->handler_runner->continue_lock();
        if(!lock)
          return;

        if(ec != asio::error::operation_aborted)
          this->accept();

        if(!ec) {
          asio::ip::tcp::no_delay option(true);
          error_code ec;
          session->connection->socket->lowest_layer().set_option(option, ec);

          session->connection->set_timeout(config.timeout_request);
          session->connection->socket->async_handshake(asio::ssl::stream_base::server, [this, session](const error_code &ec) {
            session->connection->cancel_timeout();
            auto lock = session->connection->handler_runner->continue_lock();
            if(!lock)
              return;
            if(!ec)
              this->read_request_and_content(session);
            else if(this->on_error)
              this->on_error(session->request, ec);
          });
        }
        else if(this->on_error)
          this->on_error(session->request, ec);
      });
    }

static void server_loop(void) {
    int rc;

    chip_unipro_init();

    switch_set_local_dev_id(NULL, SWITCH_PORT_ID, LOCAL_DEV_ID);

    dbgprint("Wait for peer...\n");
    chip_reset_before_ready();

    rc = svc_wait_for_peer_ready();
    if (rc) {
        return;
    }

    switch_if_dev_id_set(NULL, PEER_PORT_ID, PEER_DEV_ID);

    create_connection(&conn[0]);
    dbgprint("Control port connected\n");

    gb_control();
    gbboot_process();
#ifdef _SIMULATION
    if (stage_to_load == FFFF_ELEMENT_STAGE_2_FW)
        chip_enter_hibern8_server();
#endif
}

static connection_t *_accept_connection(void)
{
    int sock;
    connection_t *con;
    char *ip;
    int serversock; 

    serversock = wait_for_serversock(100);
    if(serversock < 0)
        return NULL;

    /* malloc enough room for a full IP address (including ipv6) */
    ip = (char *)malloc(MAX_ADDR_LEN);

    sock = sock_accept(serversock, ip, MAX_ADDR_LEN);
    if (sock >= 0) {
        con = create_connection(sock, serversock, ip);

        return con;
    }

    if (!sock_recoverable(sock_error()))
        WARN2("accept() failed with error %d: %s", sock_error(), strerror(sock_error()));
    
    free(ip);

    return NULL;
}

/**
   \details Initialize mapistore named properties MySQL backend

   \param mem_ctx pointer to the memory context
   \param lp_ctx pointer to the loadparm context
   \param nprops_ctx pointer on pointer to the namedprops context to
   return

   \return MAPISTORE_SUCCESS on success, otherwise MAPISTORE error
 */
enum mapistore_error mapistore_namedprops_mysql_init(TALLOC_CTX *mem_ctx,
						     struct loadparm_context *lp_ctx,
						     struct namedprops_context **nprops_ctx)
{
	enum mapistore_error		retval;
	struct namedprops_context	*nprops = NULL;
	struct namedprops_mysql_params	parms;
	MYSQL				*conn = NULL;

	/* Sanity checks */
	MAPISTORE_RETVAL_IF(!lp_ctx, MAPISTORE_ERR_INVALID_PARAMETER, NULL);
	MAPISTORE_RETVAL_IF(!nprops_ctx, MAPISTORE_ERR_INVALID_PARAMETER, NULL);

	/* Retrieve smb.conf arguments */
	retval = mapistore_namedprops_mysql_parameters(lp_ctx, &parms);
	if (retval != MAPISTORE_SUCCESS) {
		DEBUG(0, ("[%s:%d] ERROR: parsing MySQL named properties "
			  "parametric option failed with %s\n",
			  __FUNCTION__, __LINE__, mapistore_errstr(retval)));
		MAPISTORE_RETVAL_ERR(retval, NULL);
	}

	/* Establish MySQL connection */
	if (parms.sock) {
		// FIXME
		DEBUG(0, ("Not implemented connect through unix socket to mysql"));
	} else {
		char *connection_string = connection_string_from_parameters(mem_ctx, &parms);
		MAPISTORE_RETVAL_IF(!connection_string, MAPISTORE_ERR_NOT_INITIALIZED, NULL);
		create_connection(connection_string, &conn);
	}
	MAPISTORE_RETVAL_IF(!conn, MAPISTORE_ERR_NOT_INITIALIZED, NULL);

	/* Initialize the database */
	if (!is_schema_created(conn) || is_database_empty(conn)) {
		retval = initialize_database(conn, parms.data);
		MAPISTORE_RETVAL_IF(retval != MAPISTORE_SUCCESS, retval, NULL);
	}

	/* Create context */
	nprops = talloc_zero(mem_ctx, struct namedprops_context);
	MAPISTORE_RETVAL_IF(!nprops, MAPISTORE_ERR_NO_MEMORY, NULL);

	nprops->backend_type = NAMEDPROPS_BACKEND_MYSQL;

	nprops->create_id = create_id;
	nprops->get_mapped_id = get_mapped_id;
	nprops->get_nameid = get_nameid;
	nprops->get_nameid_type = get_nameid_type;
	nprops->next_unused_id = next_unused_id;
	nprops->transaction_commit = transaction_commit;
	nprops->transaction_start = transaction_start;

	nprops->data = conn;
	talloc_set_destructor(nprops, mapistore_namedprops_mysql_destructor);

	*nprops_ctx = nprops;
	return MAPISTORE_SUCCESS;
}

void cacheEngine_init(char* cache_server_ip, char* cache_server_mac,char* controller_ip_address, unsigned short controller_port)
{
	fprintf(stderr,"Opening socket to talk with controller\n");
	socket_to_controller=create_connection(controller_ip_address, controller_port);
	if (socket_to_controller<0)
		error("ERROR opening socket to the controller\n");
		
	sendWelcomeMsgToController(cache_server_mac, cache_server_ip);

}

gint
main (gint argc,
      gchar *argv[])
{
  g_autoptr(GDBusConnection) connection = NULL;
  g_autoptr(IpcGitService) service = NULL;
  g_autoptr(GOutputStream) stdout_stream = NULL;
  g_autoptr(GInputStream) stdin_stream = NULL;
  g_autoptr(GIOStream) stream = NULL;
  g_autoptr(GMainLoop) main_loop = NULL;
  g_autoptr(GError) error = NULL;

  g_set_prgname ("gnome-builder-git");
  g_set_application_name ("gnome-builder-git");

  prctl (PR_SET_PDEATHSIG, SIGTERM);

  signal (SIGPIPE, SIG_IGN);

  ggit_init ();

  g_log_set_handler (NULL, G_LOG_LEVEL_MASK, log_func, NULL);

  if (!g_unix_set_fd_nonblocking (STDIN_FILENO, TRUE, &error) ||
      !g_unix_set_fd_nonblocking (STDOUT_FILENO, TRUE, &error))
    goto error;

  main_loop = g_main_loop_new (NULL, FALSE);
  stdin_stream = g_unix_input_stream_new (STDIN_FILENO, FALSE);
  stdout_stream = g_unix_output_stream_new (STDOUT_FILENO, FALSE);
  stream = g_simple_io_stream_new (stdin_stream, stdout_stream);

  if (!(connection = create_connection (stream, main_loop, &error)))
    goto error;

  service = ipc_git_service_impl_new ();

  if (!g_dbus_interface_skeleton_export (G_DBUS_INTERFACE_SKELETON (service),
                                         connection,
                                         "/org/gnome/Builder/Git",
                                         &error))
    goto error;

  g_dbus_connection_start_message_processing (connection);
  g_main_loop_run (main_loop);

  return EXIT_SUCCESS;

error:
  if (error != NULL)
    g_printerr ("%s\n", error->message);

  return EXIT_FAILURE;
}

static void
nmt_add_connection_show (NmtNewtForm *form)
{
	NmtAddConnectionPrivate *priv = NMT_ADD_CONNECTION_GET_PRIVATE (form);

	if (priv->single_type) {
		nmt_newt_listbox_set_active (priv->listbox, 0);
		create_connection (NMT_NEWT_WIDGET (priv->listbox), g_object_ref (form));
	} else
		NMT_NEWT_FORM_CLASS (nmt_add_connection_parent_class)->show (form);
}

void ConnectionPool::recycle_connections()
{
  // The recycler periodically recycles the connections so that any new nodes
  // in the upstream proxy cluster get used reasonably soon after they are
  // active.  To avoid mucking around with variable length waits, the
  // algorithm waits for a fixed period (one second) then recycles connections
  // that are due to be recycled.

  while (!_terminated)
  {
#ifdef UNIT_TEST
    // A smaller pause here is much faster
    sleep(0.1);
#else
    sleep(1);
#endif

    int now = time(NULL);

    // Walk the vector of connections.  This is safe to do without the lock
    // because the vector is immutable.
    for (size_t ii = 0; ii < _tp_hash.size(); ++ii)
    {
      if (_tp_hash[ii].tp == NULL)
      {
        // This slot is empty, so try to populate it now.
        create_connection(ii);
      }
      else if ((_tp_hash[ii].connected) &&
               (_tp_hash[ii].recycle_time != 0) &&
               (now >= _tp_hash[ii].recycle_time))
      {
        // This slot is due to be recycled, so quiesce the existing
        // connection and create a new one.
        TRC_STATUS("Recycle TCP connection slot %d", ii);
        quiesce_connection(ii);
        create_connection(ii);
      }
    }
  }
}