static void tcp_receive_message (IP_CX *cxp)
{
	TCP_MSG		*recv_msg;
	TCP_DATA	*dp;

	if ((dp = cxp->sproto) == NULL || !cxp->fd) {
		warn_printf ("tcp_receive_message: no TCP context for fd:%u!", cxp->fd);
		return;
	}
	ctrc_printd (TCPS_ID, TCPS_RX_MSG, &cxp->fd, sizeof (cxp->fd));
	if (!dp->recv_msg) {

		/* Prepare for receiving messages */
		dp->recv_msg = xmalloc (sizeof (TCP_MSG));
		if (!dp->recv_msg) {
			tcp_cleanup_ctx (cxp);
			return;
		}
		dp->recv_msg->size = dp->recv_msg->used = 0;
		dp->recv_msg->buffer = NULL;
	}

	if (tcp_receive_message_fragment (cxp->fd, dp->recv_msg) == -1) {
		tcp_cleanup_ctx (cxp);
		return;
	}
	if (dp->recv_msg->used == dp->recv_msg->size) {
		recv_msg = dp->recv_msg;
		dp->recv_msg = NULL;
		ctrc_printd (TCPS_ID, TCPS_NEW_MSG, &cxp->fd, sizeof (cxp->fd));
		cxp->stream_cb->on_new_message (cxp, recv_msg->buffer, recv_msg->size);
		xfree (recv_msg->buffer);
		xfree (recv_msg);
	}
}

static int sfw_be_new_change (RemReader_t      *rrp,
			      Change_t         *cp,
			      HCI              hci,
			      SequenceNumber_t *snr)
{
	CCREF	*rp;

	ARG_NOT_USED (snr)

	ctrc_printd (RTPS_ID, RTPS_SFW_BE_NEW, &rrp, sizeof (rrp));
	prof_start (rtps_bw_new);

	RR_SIGNAL (rrp, "BE-NewChange");

#ifdef RTPS_MARKERS
	if (rrp->rr_writer->endpoint.mark_newch)
		rtps_marker_notify (rrp->rr_writer->endpoint.endpoint, EM_NEW_CHANGE, "sfw_be_new_change");
#endif
	rp = change_enqueue (rrp, cp, hci, CS_UNSENT);
	if (!rp)
		return (0);

	rp->ack_req = 1;
	rrp->rr_unacked++;
	if (!rrp->rr_unsent_changes) {
		rrp->rr_unsent_changes = rp;
		proxy_activate (&rrp->proxy);
	}
	NEW_RR_TSTATE (rrp, RRTS_PUSHING, 0);
	CACHE_CHECK (&rrp->rr_writer->endpoint, "sfw_be_new_change");
	prof_stop (rtps_bw_new, 1);
	return (1);
}

static void sfw_be_start (RemReader_t *rrp)
{
	ctrc_printd (RTPS_ID, RTPS_SFW_BE_START, &rrp, sizeof (rrp));
	prof_start (rtps_bw_start);

	RR_SIGNAL (rrp, "BE-Start");
	NEW_RR_CSTATE (rrp, RRCS_INITIAL, 1);
	NEW_RR_CSTATE (rrp, RRCS_READY, 0);
	NEW_RR_TSTATE (rrp, RRTS_IDLE, 1);
	rrp->rr_nack_timer = NULL;

#ifdef RTPS_MARKERS
	if (rrp->rr_writer->endpoint.mark_start)
		rtps_marker_notify (rrp->rr_writer->endpoint.endpoint, EM_START, "sfw_be_start");
#endif

	/* Add existing cache entries to reader locator/proxy queue. */
	
	hc_replay (rrp->rr_writer->endpoint.endpoint->cache,
					proxy_add_change, (uintptr_t) rrp);

	if ((rrp->rr_unsent_changes = LIST_HEAD (rrp->rr_changes)) != NULL &&
	    rrp->rr_writer->endpoint.push_mode) {
		NEW_RR_TSTATE (rrp, RRTS_PUSHING, 0);
		proxy_activate (&rrp->proxy);
	}
	prof_stop (rtps_bw_start, 1);
	CACHE_CHECK (&rrp->rr_writer->endpoint, "sfw_be_start");
}

DDS_ReturnCode_t DDS_Subscriber_set_qos (DDS_Subscriber sp,
					 DDS_SubscriberQos *qos)
{
	Domain_t		*dp;
	DDS_ReturnCode_t	ret;

	ctrc_printd (DCPS_ID, DCPS_S_S_QOS, &sp, sizeof (sp));
	ctrc_contd (&qos, sizeof (qos));
	ctrc_endd ();

	if (!subscriber_ptr (sp, &ret))
		return (ret);

	dp = domain_ptr (sp->domain, 1, &ret);
	if (!dp)
		return (ret);

	if (qos == DDS_PUBLISHER_QOS_DEFAULT)
		qos = &sp->domain->def_subscriber_qos;
	else if (!qos_valid_subscriber_qos (qos)) {
		ret = DDS_RETCODE_BAD_PARAMETER;
		goto done;
	}
	ret = qos_subscriber_update (&sp->qos, qos);
	if (ret == DDS_RETCODE_OK)
		sl_walk (&sp->domain->participant.p_endpoints, update_reader_qos, sp);

    done:
	lock_release (dp->lock);
	return (ret);
}

DDS_ReturnCode_t DDS_Subscriber_set_listener (DDS_Subscriber sp,
					      DDS_SubscriberListener *listener,
					      DDS_StatusMask mask)
{
	Domain_t		*dp;
	DDS_ReturnCode_t	ret;

	ctrc_printd (DCPS_ID, DCPS_S_S_LIS, &sp, sizeof (sp));
	ctrc_contd (&listener, sizeof (listener));
	ctrc_contd (&mask, sizeof (mask));
	ctrc_endd ();

	if (!subscriber_ptr (sp, &ret))
		return (ret);

	dp = domain_ptr (sp->domain, 1, &ret);
	if (!dp)
		return (ret);

	dcps_update_listener ((Entity_t *) sp, &dp->lock,
			      &sp->mask, &sp->listener,
			      mask, listener);
	lock_release (dp->lock);
	return (DDS_RETCODE_OK);
}

DDS_ReturnCode_t DDS_Subscriber_enable (DDS_Subscriber sp)
{
	Domain_t		*dp;
	DDS_ReturnCode_t	ret;

	ctrc_printd (DCPS_ID, DCPS_S_ENABLE, &sp, sizeof (sp));

	if (!subscriber_ptr (sp, &ret))
		return (ret);

	dp = domain_ptr (sp->domain, 1, &ret);
	if (!dp)
		return (ret);

	if ((dp->participant.p_flags & EF_ENABLED) == 0) {
		lock_release (dp->lock);
		return (DDS_RETCODE_NOT_ENABLED);
	}
	if ((sp->entity.flags & EF_ENABLED) == 0) {

		/* ... todo ... */

		sp->entity.flags |= EF_ENABLED | EF_NOT_IGNORED;
	}
	lock_release (dp->lock);
	return (DDS_RETCODE_OK);
}

DDS_StatusCondition DDS_Subscriber_get_statuscondition (DDS_Subscriber sp)
{
	Domain_t		*dp;
	StatusCondition_t	*scp;

	ctrc_printd (DCPS_ID, DCPS_S_G_SCOND, &sp, sizeof (sp));

	if (!subscriber_ptr (sp, NULL))
		return (NULL);

	dp = domain_ptr (sp->domain, 1, NULL);
	if (!dp)
		return (NULL);

	scp = sp->condition;
	if (!scp) {
		scp = dcps_new_status_condition ();
		if (!scp)
			return (NULL);

		scp->entity = (Entity_t *) sp;
		sp->condition = scp;
	}
	lock_release (dp->lock);
	return ((DDS_StatusCondition) scp);
}

DDS_DataWriterListener *DDS_DataWriter_get_listener (DDS_DataWriter wp)
{
	ctrc_printd (DCPS_ID, DCPS_DW_G_LIS, &wp, sizeof (wp));

	if (!writer_ptr (wp, 0, NULL))
		return (NULL);

	return (&wp->w_listener);
}

DDS_TopicListener *DDS_Topic_get_listener (DDS_Topic tp)
{
	ctrc_printd (DCPS_ID, DCPS_T_G_LIS, &tp, sizeof (tp));

	if (!topic_ptr (tp, 0, NULL))
		return (NULL);

	return (&tp->listener);
}

DDS_InstanceHandle_t DDS_Subscriber_get_instance_handle (DDS_Subscriber sp)
{
	ctrc_printd (DCPS_ID, DCPS_S_G_HANDLE, &sp, sizeof (sp));

	if (!subscriber_ptr (sp, NULL))
		return (0);

	return (sp->entity.handle);
}

static void tcp_wait_connect_complete (SOCKET fd, short revents, void *arg)
{
	TCP_CON_REQ_ST	*p = (TCP_CON_REQ_ST *) arg;
	IP_CX		*cxp = p->cxp;
	socklen_t	s;
	int		err, r;
	socklen_t	sz;

	ctrc_begind (TCPS_ID, TCPS_WCON_EV, &fd, sizeof (fd));
	ctrc_contd (&revents, sizeof (revents));
	ctrc_endd ();

	trace_poll_events (fd, revents, arg);

	trc_con2 ("tcp_wait_connect_complete(cp=%p, cxp=%p);\r\n", (void *) p, (void *) cxp);
	p = tcp_clear_pending_connect (p);
	do {
		if ((revents & (POLLERR | POLLNVAL)) != 0) {
			sz = sizeof (err);
			r = getsockopt (cxp->fd, SOL_SOCKET, SO_ERROR, &err, &sz);
			if (r == -1 || err)  {
				log_printf (RTPS_ID, 0, "POLLERR | POLLNVAL [%d]: %d %s\r\n", cxp->fd, err, strerror (err));
				tcp_cleanup_ctx (cxp);
				break;
			}
		}
		if ((revents & POLLHUP) != 0) {
			tcp_cleanup_ctx (cxp);
			break;
		}
		if ((revents & POLLOUT) != 0) {
			s = sizeof (err);
			r = getsockopt (cxp->fd, SOL_SOCKET, SO_ERROR, &err, &s);
			if (r || err) {
				if (r)
					perror ("cc_control: getsockopt(SOL_SOCKET/SO_ERROR)");
				tcp_cleanup_ctx (cxp);
				break;
			}
		}
		log_printf (RTPS_ID, 0, "TCP: server connection established [%d]\r\n", cxp->fd);
		cxp->cx_state = CXS_OPEN;
		sock_fd_remove_socket (cxp->fd);
		sock_fd_add_socket (cxp->fd,
				    POLLIN | POLLPRI | POLLHUP | POLLNVAL,
				    tcp_socket_activity,
				    cxp, "DDS.TCP-H");
		ctrc_printd (TCPS_ID, TCPS_CON_COMPL, &fd, sizeof (fd));
		cxp->stream_cb->on_connected (cxp);
	}
	while (0);
	if (p)
		tcp_do_connect (p);
}

DDS_Publisher DDS_DataWriter_get_publisher (DDS_DataWriter wp)
{
	Publisher_t	*up;

	ctrc_printd (DCPS_ID, DCPS_DW_G_PUB, &wp, sizeof (wp));

	if (!writer_ptr (wp, 0, NULL))
		return (NULL);

	up = wp->w_publisher;
	return (up);
}

DDS_InstanceHandle_t DDS_Topic_get_instance_handle (DDS_Topic tp)
{
	DDS_InstanceHandle_t	h;

	ctrc_printd (DCPS_ID, DCPS_T_G_HANDLE, &tp, sizeof (tp));

	if (!topic_ptr (tp, 1, NULL))
		return (0);

	h = tp->entity.handle;
	lock_release (tp->lock);
	return (h);
}

DDS_StatusMask DDS_Topic_get_status_changes (DDS_Topic tp)
{
	DDS_StatusMask	m;

	ctrc_printd (DCPS_ID, DCPS_T_G_STAT, &tp, sizeof (tp));

	if (!topic_ptr (tp, 1, NULL))
		return (0);

	m = tp->status;
	lock_release (tp->lock);
	return (m);
}

static void tcp_cleanup_ctx (IP_CX *cxp)
{
	ctrc_printd (TCPS_ID, TCPS_CLEAN, &cxp->fd, sizeof (cxp->fd));
	if (cxp->stream_cb && cxp->stream_cb->on_close)
		cxp->stream_cb->on_close (cxp); /* This fd most probably became unusable. */
	else if (cxp->fd_owner) {
		close (cxp->fd);
		sock_fd_remove_socket (cxp->fd);
		cxp->cx_state = CXS_CLOSED;
		cxp->fd_owner = 0;
		cxp->fd = 0;
	}
}

DDS_InstanceHandle_t DDS_DataWriter_get_instance_handle (DDS_DataWriter wp)
{
	DDS_InstanceHandle_t	h;

	ctrc_printd (DCPS_ID, DCPS_DW_G_HANDLE, &wp, sizeof (wp));

	if (!writer_ptr (wp, 1, NULL))
		return (0);

	h = wp->w_handle;
	lock_release (wp->w_lock);
	return (h);
}

DDS_DomainId_t DDS_DomainParticipant_get_domain_id (DDS_DomainParticipant dp)
{
	DDS_DomainId_t		id;
	DDS_ReturnCode_t	ret;

	ctrc_printd (DCPS_ID, DCPS_DP_G_ID, &p, sizeof (p));
	if (!domain_ptr (dp, 1, &ret))
		return (0);

	id = dp->domain_id;
	lock_release (dp->lock);
	return (id);
}

DDS_StatusMask DDS_DataWriter_get_status_changes (DDS_DataWriter wp)
{
	DDS_StatusMask	m;
	
	ctrc_printd (DCPS_ID, DCPS_DW_G_STAT, &wp, sizeof (wp));

	if (!writer_ptr (wp, 1, NULL))
		return (0);

	m = wp->w_status;
	lock_release (wp->w_lock);
	return (m);
}

DDS_InstanceHandle_t DDS_DomainParticipant_get_instance_handle (DDS_DomainParticipant dp)
{
	DDS_InstanceHandle_t	handle;

	ctrc_printd (DCPS_ID, DCPS_DP_G_HANDLE, &dp, sizeof (dp));

	if (!domain_ptr (dp, 1, NULL))
		return (DDS_HANDLE_NIL);

	handle = dp->participant.p_handle;
	lock_release (dp->lock);
	return (handle);
}

DDS_StatusMask DDS_DomainParticipant_get_status_changes (DDS_DomainParticipant dp)
{
	DDS_StatusMask	m;
	
	ctrc_printd (DCPS_ID, DCPS_DP_G_STAT, &dp, sizeof (dp));
	if (!domain_ptr (dp, 1, NULL))
		return (0);

	/* No status mask on participant for now! */
	m = 0U;

	lock_release (dp->lock);
	return (m);
}