/* Decrement the key's reference counter, and when the counter reach zero,
 * destroy the key.
 *
 * Note: MUST NOT CALL THIS FUNCTION WHILE HOLDING ioqueue's LOCK.
 */
static void decrement_counter(pj_ioqueue_key_t *key)
{
    pj_lock_acquire(key->ioqueue->lock);
    pj_mutex_lock(key->ioqueue->ref_cnt_mutex);
    --key->ref_count;
    if (key->ref_count == 0) {

	pj_assert(key->closing == 1);
	pj_gettickcount(&key->free_time);
	key->free_time.msec += PJ_IOQUEUE_KEY_FREE_DELAY;
	pj_time_val_normalize(&key->free_time);

	pj_list_erase(key);
	pj_list_push_back(&key->ioqueue->closing_list, key);

    }
    pj_mutex_unlock(key->ioqueue->ref_cnt_mutex);
    pj_lock_release(key->ioqueue->lock);
}

PJ_DEF(int) pj_ioqueue_recvfrom( pj_ioqueue_t *ioque,
				 pj_ioqueue_key_t *key,
				 void *buffer,
				 pj_size_t buflen,
				 pj_sockaddr_t *addr,
				 int *addrlen)
{
    pj_mutex_lock(ioque->mutex);

    key->op |= PJ_IOQUEUE_OP_RECV_FROM;
    key->rd_buf = buffer;
    key->rd_buflen = buflen;
    key->rmt_addr = addr;
    key->rmt_addrlen = addrlen;
    PJ_FD_SET(key->fd, &ioque->rfdset);

    pj_mutex_unlock(ioque->mutex);
    return PJ_IOQUEUE_PENDING;
}

/* API: Register device change observer. */
PJ_DEF(pj_status_t) pjmedia_aud_dev_set_observer_cb(pjmedia_aud_dev_observer_callback cb)
{
    pj_status_t status;

    status = pj_mutex_lock(aud_subsys.dev_observer.lock);
    if (status != PJ_SUCCESS) {
        PJ_LOG(5, (THIS_FILE, "Could not acquire audio device change lock"));
        return status;
    }

    aud_subsys.dev_observer.cb = cb;

    status = pj_mutex_unlock(aud_subsys.dev_observer.lock);
    if (status != PJ_SUCCESS) {
        PJ_LOG(5, (THIS_FILE, "Could not release audio device change lock"));
    }

    return status;
}

/* Event worker thread function. */
static int event_worker_thread(void *arg)
{
    pjmedia_event_mgr *mgr = (pjmedia_event_mgr *)arg;

    while (1) {
	/* Wait until there is an event. */
        pj_sem_wait(mgr->sem);

        if (mgr->is_quitting)
            break;

        pj_mutex_lock(mgr->mutex);
        event_mgr_distribute_events(mgr, &mgr->ev_queue,
                                    &mgr->th_next_sub, PJ_TRUE);
        pj_mutex_unlock(mgr->mutex);
    }

    return 0;
}

/*
 * Allocate a new Speex codec instance.
 */
static pj_status_t spx_alloc_codec( pjmedia_codec_factory *factory, 
				    const pjmedia_codec_info *id,
				    pjmedia_codec **p_codec)
{
    pjmedia_codec *codec;
    struct spx_private *spx;

    PJ_ASSERT_RETURN(factory && id && p_codec, PJ_EINVAL);
    PJ_ASSERT_RETURN(factory == &spx_factory.base, PJ_EINVAL);


    pj_mutex_lock(spx_factory.mutex);

    /* Get free nodes, if any. */
    if (!pj_list_empty(&spx_factory.codec_list)) {
	codec = spx_factory.codec_list.next;
	pj_list_erase(codec);
    } else {
	codec = PJ_POOL_ZALLOC_T(spx_factory.pool, pjmedia_codec);
	PJ_ASSERT_RETURN(codec != NULL, PJ_ENOMEM);
	codec->op = &spx_op;
	codec->factory = factory;
	codec->codec_data = pj_pool_alloc(spx_factory.pool,
					  sizeof(struct spx_private));
    }

    pj_mutex_unlock(spx_factory.mutex);

    spx = (struct spx_private*) codec->codec_data;
    spx->enc = NULL;
    spx->dec = NULL;

    if (id->clock_rate <= 8000)
	spx->param_id = PARAM_NB;
    else if (id->clock_rate <= 16000)
	spx->param_id = PARAM_WB;
    else
	spx->param_id = PARAM_UWB;

    *p_codec = codec;
    return PJ_SUCCESS;
}

/*
 * Find codecs by the unique codec identifier. This function will find
 * all codecs that match the codec identifier prefix. For example, if
 * "L16" is specified, then it will find "L16/8000/1", "L16/16000/1",
 * and so on, up to the maximum count specified in the argument.
 */
PJ_DEF(pj_status_t) pjmedia_vid_codec_mgr_find_codecs_by_id(
                                     pjmedia_vid_codec_mgr *mgr,
				     const pj_str_t *codec_id,
				     unsigned *count,
				     const pjmedia_vid_codec_info *p_info[],
				     unsigned prio[])
{
    unsigned i, found = 0;

    PJ_ASSERT_RETURN(codec_id && count && *count, PJ_EINVAL);

    if (!mgr) mgr = def_vid_codec_mgr;
    PJ_ASSERT_RETURN(mgr, PJ_EINVAL);

    pj_mutex_lock(mgr->mutex);

    for (i=0; i<mgr->codec_cnt; ++i) {

	if (codec_id->slen == 0 ||
	    pj_strnicmp2(codec_id, mgr->codec_desc[i].id, 
			 codec_id->slen) == 0) 
	{

	    if (p_info)
		p_info[found] = &mgr->codec_desc[i].info;
	    if (prio)
		prio[found] = mgr->codec_desc[i].prio;

	    ++found;

	    if (found >= *count)
		break;
	}

    }

    pj_mutex_unlock(mgr->mutex);

    *count = found;

    return found ? PJ_SUCCESS : PJ_ENOTFOUND;
}

/*
 * Modify codec settings.
 */
static pj_status_t  codec_modify( pjmedia_codec *codec, 
				  const pjmedia_codec_param *attr )
{
    struct opus_data *opus_data = (struct opus_data *)codec->codec_data;

    pj_mutex_lock (opus_data->mutex);

    TRACE_((THIS_FILE, "%s:%d: - TRACE", __FUNCTION__, __LINE__));

    /* Set bitrate */
    opus_data->cfg.bit_rate = attr->info.avg_bps;
    opus_encoder_ctl(opus_data->enc, OPUS_SET_BITRATE(attr->info.avg_bps));
    /* Set VAD */
    opus_encoder_ctl(opus_data->enc, OPUS_SET_DTX(attr->setting.vad ? 1 : 0));
    /* Set PLC */
    opus_encoder_ctl(opus_data->enc,
    		     OPUS_SET_INBAND_FEC(attr->setting.plc ? 1 : 0));

    pj_mutex_unlock (opus_data->mutex);
    return PJ_SUCCESS;
}

/*
 * Release read lock.
 *
 */
PJ_DEF(pj_status_t) pj_rwmutex_unlock_read(pj_rwmutex_t *mutex)
{
    pj_status_t status;

    PJ_ASSERT_RETURN(mutex, PJ_EINVAL);

    status = pj_mutex_lock(mutex->read_lock);
    if (status != PJ_SUCCESS) {
	pj_assert(!"This pretty much is unexpected");
	return status;
    }

    pj_assert(mutex->reader_count >= 1);

    --mutex->reader_count;
    if (mutex->reader_count == 0)
	pj_sem_post(mutex->write_lock);

    status = pj_mutex_unlock(mutex->read_lock);
    return status;
}

static pj_status_t l16_dealloc_codec(pjmedia_codec_factory *factory, 
				     pjmedia_codec *codec )
{
    struct l16_data *data;

    PJ_ASSERT_RETURN(factory && codec, PJ_EINVAL);
    PJ_ASSERT_RETURN(factory==&l16_factory.base, PJ_EINVAL);

    /* Lock mutex. */
    pj_mutex_lock(l16_factory.mutex);

    /* Just release codec data pool */
    data = (struct l16_data*) codec->codec_data;
    pj_assert(data);
    pj_pool_release(data->pool);

    /* Unlock mutex. */
    pj_mutex_unlock(l16_factory.mutex);

    return PJ_SUCCESS;
}

/*
 * Free codec.
 */
static pj_status_t spx_dealloc_codec( pjmedia_codec_factory *factory, 
				      pjmedia_codec *codec )
{
    struct spx_private *spx;

    PJ_ASSERT_RETURN(factory && codec, PJ_EINVAL);
    PJ_ASSERT_RETURN(factory == &spx_factory.base, PJ_EINVAL);

    /* Close codec, if it's not closed. */
    spx = (struct spx_private*) codec->codec_data;
    if (spx->enc != NULL || spx->dec != NULL) {
	spx_codec_close(codec);
    }

    /* Put in the free list. */
    pj_mutex_lock(spx_factory.mutex);
    pj_list_push_front(&spx_factory.codec_list, codec);
    pj_mutex_unlock(spx_factory.mutex);

    return PJ_SUCCESS;
}

/* If this timer callback is called, it means subscriber hasn't refreshed its
 * subscription on-time. Set the state to terminated. This will also send
 * NOTIFY with Subscription-State set to terminated.
 */
static void uas_expire_timer_cb( pj_timer_heap_t *timer_heap, pj_timer_entry *entry)
{
    pjsip_event_sub *sub = entry->user_data;
    pj_str_t reason = { "timeout", 7 };

    PJ_LOG(4,(THIS_FILE, "event_sub%p (%s): UAS subscription expired!", 
			 sub, state[sub->state].ptr));

    pj_mutex_lock(sub->mutex);
    sub->timer.id = 0;

    if (sub->cb.on_sub_terminated && sub->state!=PJSIP_EVENT_SUB_STATE_TERMINATED) {
	/* Notify application, but prevent app from destroying the sub. */
	++sub->pending_tsx;
	(*sub->cb.on_sub_terminated)(sub, &reason);
	--sub->pending_tsx;
    }
    //pjsip_event_sub_notify( sub, PJSIP_EVENT_SUB_STATE_TERMINATED, 
    //			    &reason, NULL);
    pj_mutex_unlock(sub->mutex);

}

/*
 * Allocate a new SILK codec instance.
 */
static pj_status_t silk_alloc_codec(pjmedia_codec_factory *factory,
				    const pjmedia_codec_info *id,
				    pjmedia_codec **p_codec)
{
    pjmedia_codec *codec;
    struct silk_private *silk;

    PJ_ASSERT_RETURN(factory && id && p_codec, PJ_EINVAL);
    PJ_ASSERT_RETURN(factory == &silk_factory.base, PJ_EINVAL);


    pj_mutex_lock(silk_factory.mutex);

    /* Get free nodes, if any. */
    if (!pj_list_empty(&silk_factory.codec_list)) {
		codec = silk_factory.codec_list.next;
		pj_list_erase(codec);
    } else {
		codec = PJ_POOL_ZALLOC_T(silk_factory.pool, pjmedia_codec);
		PJ_ASSERT_RETURN(codec != NULL, PJ_ENOMEM);
		codec->op = &silk_op;
		codec->factory = factory;
		codec->codec_data = pj_pool_alloc(silk_factory.pool,
						  sizeof(struct silk_private));
    }

    pj_mutex_unlock(silk_factory.mutex);

    silk = (struct silk_private*) codec->codec_data;
    silk->enc_ready = PJ_FALSE;
    silk->dec_ready = PJ_FALSE;
    silk->param_id = silk_get_type_for_clock_rate(id->clock_rate);

    /* Create pool for codec instance */
    silk->pool = pjmedia_endpt_create_pool(silk_factory.endpt, "silkcodec", 512, 512);

    *p_codec = codec;
    return PJ_SUCCESS;
}

/*!
 * \internal
 * \brief Find the request tdata to get the serializer it used.
 * \since 14.0.0
 *
 * \param rdata The incoming message.
 *
 * \retval serializer on success.
 * \retval NULL on error or could not find the serializer.
 */
static struct ast_taskprocessor *find_request_serializer(pjsip_rx_data *rdata)
{
	struct ast_taskprocessor *serializer = NULL;
	pj_str_t tsx_key;
	pjsip_transaction *tsx;

	pjsip_tsx_create_key(rdata->tp_info.pool, &tsx_key, PJSIP_ROLE_UAC,
		&rdata->msg_info.cseq->method, rdata);

	tsx = pjsip_tsx_layer_find_tsx(&tsx_key, PJ_TRUE);
	if (!tsx) {
		ast_debug(1, "Could not find %.*s transaction for %d response.\n",
			(int) pj_strlen(&rdata->msg_info.cseq->method.name),
			pj_strbuf(&rdata->msg_info.cseq->method.name),
			rdata->msg_info.msg->line.status.code);
		return NULL;
	}

	if (tsx->last_tx) {
		const char *serializer_name;

		serializer_name = tsx->last_tx->mod_data[distributor_mod.id];
		if (!ast_strlen_zero(serializer_name)) {
			serializer = ast_taskprocessor_get(serializer_name, TPS_REF_IF_EXISTS);
			if (serializer) {
				ast_debug(3, "Found serializer %s on transaction %s\n",
						serializer_name, tsx->obj_name);
			}
		}
	}

#ifdef HAVE_PJ_TRANSACTION_GRP_LOCK
	pj_grp_lock_release(tsx->grp_lock);
#else
	pj_mutex_unlock(tsx->mutex);
#endif

	return serializer;
}

static pj_status_t event_mgr_distribute_events(pjmedia_event_mgr *mgr,
                                               event_queue *ev_queue,
                                               esub **next_sub,
                                               pj_bool_t rls_lock)
{
    pj_status_t err = PJ_SUCCESS;
    esub * sub = mgr->esub_list.next;
    pjmedia_event *ev = &ev_queue->events[ev_queue->head];

    while (sub != &mgr->esub_list) {
        *next_sub = sub->next;

        /* Check if the subscriber is interested in 
         * receiving the event from the publisher.
         */
        if (sub->epub == ev->epub || !sub->epub) {
            pjmedia_event_cb *cb = sub->cb;
            void *user_data = sub->user_data;
            pj_status_t status;
            
            if (rls_lock)
                pj_mutex_unlock(mgr->mutex);

            status = (*cb)(ev, user_data);
            if (status != PJ_SUCCESS && err == PJ_SUCCESS)
	        err = status;

            if (rls_lock)
                pj_mutex_lock(mgr->mutex);
        }
	sub = *next_sub;
    }
    *next_sub = NULL;

    ev_queue->head = (ev_queue->head + 1) % MAX_EVENTS;
    ev_queue->is_full = PJ_FALSE;

    return err;
}

/*
 * Set route-set.
 */
PJ_DEF(pj_status_t) pjsip_event_sub_set_route_set( pjsip_event_sub *sub,
						   const pjsip_route_hdr *route_set )
{
    const pjsip_route_hdr *hdr;

    pj_mutex_lock(sub->mutex);

    /* Clear existing route set. */
    pj_list_init(&sub->route_set);

    /* Duplicate route headers. */
    hdr = route_set->next;
    while (hdr != route_set) {
	pjsip_route_hdr *new_hdr = pjsip_hdr_clone(sub->pool, hdr);
	pj_list_insert_before(&sub->route_set, new_hdr);
	hdr = hdr->next;
    }

    pj_mutex_unlock(sub->mutex);

    return 0;
}

pjmedia_event_unsubscribe(pjmedia_event_mgr *mgr,
                          pjmedia_event_cb *cb,
                          void *user_data,
                          void *epub)
{
    esub *sub;

    PJ_ASSERT_RETURN(cb, PJ_EINVAL);

    if (!mgr) mgr = pjmedia_event_mgr_instance();
    PJ_ASSERT_RETURN(mgr, PJ_EINVAL);

    pj_mutex_lock(mgr->mutex);
    sub = mgr->esub_list.next;
    while (sub != &mgr->esub_list) {
	esub *next = sub->next;
        if (sub->cb == cb && (sub->user_data == user_data || !user_data) &&
            (sub->epub == epub || !epub))
        {
            /* If the worker thread or pjmedia_event_publish() API is
             * in the process of distributing events, make sure that
             * its pointer to the next subscriber stays valid.
             */
            if (mgr->th_next_sub == sub)
                mgr->th_next_sub = sub->next;
            if (mgr->pub_next_sub == sub)
                mgr->pub_next_sub = sub->next;
            pj_list_erase(sub);
            pj_list_push_back(&mgr->free_esub_list, sub);
            if (user_data && epub)
                break;
        }
	sub = next;
    }
    pj_mutex_unlock(mgr->mutex);

    return PJ_SUCCESS;
}

/* Find subscription in the hash table. 
 * If found, lock the subscription before returning to caller.
 */
static pjsip_event_sub *find_sub(pjsip_rx_data *rdata)
{
    pj_str_t key;
    pjsip_role_e role;
    pjsip_event_sub *sub;
    pjsip_method *method = &rdata->msg->line.req.method;
    pj_str_t *tag;

    if (rdata->msg->type == PJSIP_REQUEST_MSG) {
	if (pjsip_method_cmp(method, &SUBSCRIBE)==0) {
	    role = PJSIP_ROLE_UAS;
	    tag = &rdata->to_tag;
	} else {
	    pj_assert(pjsip_method_cmp(method, &NOTIFY) == 0);
	    role = PJSIP_ROLE_UAC;
	    tag = &rdata->to_tag;
	}
    } else {
	if (pjsip_method_cmp(&rdata->cseq->method, &SUBSCRIBE)==0) {
	    role = PJSIP_ROLE_UAC;
	    tag = &rdata->from_tag;
	} else {
	    pj_assert(pjsip_method_cmp(method, &NOTIFY) == 0);
	    role = PJSIP_ROLE_UAS;
	    tag = &rdata->from_tag;
	}
    }
    create_subscriber_key( &key, rdata->pool, role, &rdata->call_id, tag);

    pj_mutex_lock(mgr.mutex);
    sub = pj_hash_get(mgr.ht, key.ptr, key.slen);
    if (sub)
	pj_mutex_lock(sub->mutex);
    pj_mutex_unlock(mgr.mutex);

    return sub;
}

/*
 * Free codec.
 */
static pj_status_t gsm_dealloc_codec( pjmedia_codec_factory *factory, 
				      pjmedia_codec *codec )
{
    struct gsm_data *gsm_data;
    int i;

    PJ_ASSERT_RETURN(factory && codec, PJ_EINVAL);
    PJ_ASSERT_RETURN(factory == &gsm_codec_factory.base, PJ_EINVAL);

    gsm_data = (struct gsm_data*) codec->codec_data;

    /* Close codec, if it's not closed. */
    gsm_codec_close(codec);

#if !PLC_DISABLED
    /* Clear left samples in the PLC, since codec+plc will be reused
     * next time.
     */
    for (i=0; i<2; ++i) {
	pj_int16_t frame[160];
	pjmedia_zero_samples(frame, PJ_ARRAY_SIZE(frame));
	pjmedia_plc_save(gsm_data->plc, frame);
    }
#else
    PJ_UNUSED_ARG(i);
#endif

    /* Re-init silence_period */
    pj_set_timestamp32(&gsm_data->last_tx, 0, 0);

    /* Put in the free list. */
    pj_mutex_lock(gsm_codec_factory.mutex);
    pj_list_push_front(&gsm_codec_factory.codec_list, codec);
    pj_mutex_unlock(gsm_codec_factory.mutex);

    return PJ_SUCCESS;
}

/*
 * Allocate a new OPUS codec instance.
 */
static pj_status_t opus_alloc_codec(pjmedia_codec_factory *factory,
                                    const pjmedia_codec_info *id,
                                    pjmedia_codec **p_codec) {
  pjmedia_codec *codec;
  struct opus_private *opus;

  PJ_ASSERT_RETURN(factory && id && p_codec, PJ_EINVAL);
  PJ_ASSERT_RETURN(factory == &opus_factory.base, PJ_EINVAL);

  pj_mutex_lock(opus_factory.mutex);

  /* Get free nodes, if any. */
  if (!pj_list_empty(&opus_factory.codec_list)) {
    codec = opus_factory.codec_list.next;
    pj_list_erase(codec);
  } else {
    codec = PJ_POOL_ZALLOC_T(opus_factory.pool, pjmedia_codec);
    PJ_ASSERT_RETURN(codec != NULL, PJ_ENOMEM);
    codec->op = &opus_op;
    codec->factory = factory;
    codec->codec_data =
        pj_pool_alloc(opus_factory.pool, sizeof(struct opus_private));
  }

  pj_mutex_unlock(opus_factory.mutex);

  opus = (struct opus_private *)codec->codec_data;
  opus->enc_ready = PJ_FALSE;
  opus->dec_ready = PJ_FALSE;

  /* Create pool for codec instance */
  opus->pool =
      pjmedia_endpt_create_pool(opus_factory.endpt, "opuscodec", 512, 512);

  *p_codec = codec;
  return PJ_SUCCESS;
}

/*
 * Initiate overlapped connect() operation (well, it's non-blocking actually,
 * since there's no overlapped version of connect()).
 */
PJ_DEF(pj_status_t) pj_ioqueue_connect( pj_ioqueue_key_t *key,
					const pj_sockaddr_t *addr,
					int addrlen )
{
    pj_status_t status;
    
    /* check parameters. All must be specified! */
    PJ_ASSERT_RETURN(key && addr && addrlen, PJ_EINVAL);

    /* Check if key is closing. */
    if (IS_CLOSING(key))
	return PJ_ECANCELLED;

    /* Check if socket has not been marked for connecting */
    if (key->connecting != 0)
        return PJ_EPENDING;
    
    status = pj_sock_connect(key->fd, addr, addrlen);
    if (status == PJ_SUCCESS) {
	/* Connected! */
	return PJ_SUCCESS;
    } else {
	if (status == PJ_STATUS_FROM_OS(PJ_BLOCKING_CONNECT_ERROR_VAL)) {
	    /* Pending! */
            pj_mutex_lock(key->mutex);
	    key->connecting = PJ_TRUE;
            ioqueue_add_to_set(key->ioqueue, key, WRITEABLE_EVENT);
            ioqueue_add_to_set(key->ioqueue, key, EXCEPTION_EVENT);
            pj_mutex_unlock(key->mutex);
	    return PJ_EPENDING;
	} else {
	    /* Error! */
	    return status;
	}
    }
}