tvheadend/src/serviceprobe.c

/*
 *  Output functions for fixed multicast streaming
 *  Copyright (C) 2007 Andreas <EFBFBD>man
 *
 *  This program is free software: you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation, either version 3 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <assert.h>
#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>




#include "tvhead.h"
#include "channels.h"
#include "subscriptions.h"
#include "serviceprobe.h"
#include "transports.h"
#include "streaming.h"


/* List of transports to be probed, protected with global_lock */
static struct th_transport_queue serviceprobe_queue;  
static pthread_cond_t serviceprobe_cond;

/**
 *
 */
void
serviceprobe_enqueue(th_transport_t *t)
{
  if(!transport_is_tv(t))
    return; /* Don't even consider non-tv channels */

  if(t->tht_sp_onqueue)
    return;

  if(t->tht_ch != NULL)
    return; /* Already mapped */

  t->tht_sp_onqueue = 1;
  TAILQ_INSERT_TAIL(&serviceprobe_queue, t, tht_sp_link);
  pthread_cond_signal(&serviceprobe_cond);
}

/**
 *
 */
static void *
serviceprobe_thread(void *aux)
{
  th_transport_t *t;
  th_subscription_t *s;
  int was_doing_work = 0;
  streaming_queue_t sq;
  streaming_message_t *sm;
  transport_feed_status_t status;
  int run;
  const char *err;
  channel_t *ch;

  pthread_mutex_lock(&global_lock);

  streaming_queue_init(&sq);

  while(1) {

    while((t = TAILQ_FIRST(&serviceprobe_queue)) == NULL) {

      if(was_doing_work) {
	tvhlog(LOG_INFO, "serviceprobe", "Now idle");
	was_doing_work = 0;
      }
      pthread_cond_wait(&serviceprobe_cond, &global_lock);
    }

    if(!was_doing_work) {
      tvhlog(LOG_INFO, "serviceprobe", "Starting");
      was_doing_work = 1;
    }

    s = subscription_create_from_transport(t, "serviceprobe", &sq.sq_st);

    transport_ref(t);
    pthread_mutex_unlock(&global_lock);

    run = 1;
    pthread_mutex_lock(&sq.sq_mutex);

    while(run) {

      while((sm = TAILQ_FIRST(&sq.sq_queue)) == NULL)
	pthread_cond_wait(&sq.sq_cond, &sq.sq_mutex);
      TAILQ_REMOVE(&sq.sq_queue, sm, sm_link);

      pthread_mutex_unlock(&sq.sq_mutex);

      if(sm->sm_type == SMT_TRANSPORT_STATUS) {
	status = sm->sm_code;

	switch(status) {
	case TRANSPORT_FEED_UNKNOWN:
	  break;

	case TRANSPORT_FEED_NO_INPUT:
	  err = "No data input from adapter detected";
	  run = 0;
	  break;

	case TRANSPORT_FEED_NO_DEMUXED_INPUT:
	  err = "No mux packets for this service";
	  run = 0;
	  break;

	case TRANSPORT_FEED_RAW_INPUT:
	  err = "Data received for service, "
	    "but no packets could be reassembled";
	  run = 0;
	  break;

	case TRANSPORT_FEED_NO_DESCRAMBLER:
	  err = "No descrambler available for service";
	  run = 0;
	  break;

	case TRANSPORT_FEED_NO_ACCESS:
	  err = "Access denied";
	  run = 0;
	  break;

	case TRANSPORT_FEED_VALID_PACKETS:
	  err = NULL;
	  run = 0;
	  break;
	}
      }

      streaming_msg_free(sm);
      pthread_mutex_lock(&sq.sq_mutex);
    }

    streaming_queue_clear(&sq.sq_queue);
    pthread_mutex_unlock(&sq.sq_mutex);

    pthread_mutex_lock(&global_lock);
    subscription_unsubscribe(s);

    if(t->tht_status != TRANSPORT_ZOMBIE) {

      if(err != NULL) {
	tvhlog(LOG_INFO, "serviceprobe", "%20s: skipped: %s",
	       t->tht_svcname, err);
      } else if(t->tht_ch == NULL) {
	ch = channel_find_by_name(t->tht_svcname, 1);
	transport_map_channel(t, ch);
    
	pthread_mutex_lock(&t->tht_stream_mutex);
	t->tht_config_change(t);
	pthread_mutex_unlock(&t->tht_stream_mutex);
      
	tvhlog(LOG_INFO, "serviceprobe", "\"%s\" mapped to channel \"%s\"",
	       t->tht_svcname, t->tht_svcname);
      }

      t->tht_sp_onqueue = 0;
      TAILQ_REMOVE(&serviceprobe_queue, t, tht_sp_link);
    }
    transport_unref(t);
  }
  return NULL;
}


/**
 *
 */
void 
serviceprobe_init(void)
{
  pthread_t ptid;
  pthread_cond_init(&serviceprobe_cond, NULL);
  TAILQ_INIT(&serviceprobe_queue);
  pthread_create(&ptid, NULL, serviceprobe_thread, NULL);
}