From: Jaroslav Kysela <perex@perex.cz>
Date: Tue, 12 Sep 2017 06:11:30 +0000 (+0200)
Subject: parser: move mpegts to packet parsing completely outside tsdemux
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=f43f3af241dcf2cb16cc0127647736821654fe40;p=thirdparty%2Ftvheadend.git

parser: move mpegts to packet parsing completely outside tsdemux
---

diff --git a/Makefile b/Makefile
index bbddf7ff0..80785db36 100644
--- a/Makefile
+++ b/Makefile
@@ -313,6 +313,7 @@ SRCS-2 = \
 	src/api/api_wizard.c
 
 SRCS-2 += \
+        src/parsers/message.c \
 	src/parsers/parsers.c \
 	src/parsers/bitstream.c \
 	src/parsers/parser_h264.c \
diff --git a/src/input/mpegts/dvb_psi.c b/src/input/mpegts/dvb_psi.c
index 0154887da..1bd4cced5 100644
--- a/src/input/mpegts/dvb_psi.c
+++ b/src/input/mpegts/dvb_psi.c
@@ -2428,7 +2428,7 @@ psi_parse_pmt
 
   lock_assert(&t->s_stream_mutex);
 
-  version = ptr[2] >> 1 & 0x1f;
+  version = (ptr[2] >> 1) & 0x1f;
   pcr_pid = extract_pid(ptr + 5);
   dllen   = (ptr[7] & 0xf) << 8 | ptr[8];
   
diff --git a/src/input/mpegts/dvb_psi_hbbtv.c b/src/input/mpegts/dvb_psi_hbbtv.c
index 068ba160e..ea3ae628c 100644
--- a/src/input/mpegts/dvb_psi_hbbtv.c
+++ b/src/input/mpegts/dvb_psi_hbbtv.c
@@ -26,8 +26,9 @@
 /**
  * Extract Hbbtv
  */
-void
-ts_recv_hbbtv_cb(mpegts_psi_table_t *mt, const uint8_t *buf, int len)
+htsmsg_t *
+ts_recv_hbbtv(mpegts_psi_table_t *mt, elementary_stream_t *st,
+              const uint8_t *buf, int len, int *_sect)
 {
   static const char *visibility_table[4] = {
     "none",
@@ -35,8 +36,6 @@ ts_recv_hbbtv_cb(mpegts_psi_table_t *mt, const uint8_t *buf, int len)
     "reserved",
     "all",
   };
-  elementary_stream_t *st = (elementary_stream_t *)mt->mt_opaque;
-  service_t *t = st->es_service;
   mpegts_psi_table_state_t *tst = NULL;
   int r, sect, last, ver, l, l2, l3, dllen, dlen, flags;
   uint8_t tableid = buf[0], dtag;
@@ -45,18 +44,16 @@ ts_recv_hbbtv_cb(mpegts_psi_table_t *mt, const uint8_t *buf, int len)
   uint32_t org_id;
   char title[256], name[256], location[256], *str;
   htsmsg_t *map, *apps = NULL, *titles = NULL;
-  void *bin;
-  size_t binlen;
 
   if (tableid != 0x74 || len < 16)
-    return;
+    return NULL;
   app_type = (buf[3] << 8) | buf[4];
   if (app_type & 1) /* testing */
-    return;
+    return NULL;
 
   r = dvb_table_begin(mt, buf + 3, len - 3,
                       tableid, app_type, 5, &tst, &sect, &last, &ver, 0);
-  if (r != 1) return;
+  if (r != 1) return NULL;
 
   p = buf;
   l = len;
@@ -160,28 +157,31 @@ ts_recv_hbbtv_cb(mpegts_psi_table_t *mt, const uint8_t *buf, int len)
     goto dvberr;
 
   dvb_table_end(mt, tst, sect);
+  if (_sect)
+    *_sect = sect;
+  return apps;
 
+dvberr:
+  htsmsg_destroy(apps);
+  htsmsg_destroy(titles);
+  return NULL;
+}
+
+void
+ts_recv_hbbtv_cb(mpegts_psi_table_t *mt, const uint8_t *buf, int len)
+{
+  elementary_stream_t *st = (elementary_stream_t *)mt->mt_opaque;
+  service_t *t = st->es_service;
+  int sect;
+  htsmsg_t *apps = ts_recv_hbbtv(mt, st, buf, len, &sect);
+  if (apps == NULL)
+    return;
   if (t->s_hbbtv == NULL)
     t->s_hbbtv = htsmsg_create_map();
   if (apps) {
-    snprintf(location, sizeof(location), "%d", sect);
-    htsmsg_set_msg(t->s_hbbtv, location, apps);
-    apps = NULL;
+    char buf[20];
+    snprintf(buf, sizeof(buf), "%d", sect);
+    htsmsg_set_msg(t->s_hbbtv, buf, apps);
     service_request_save(t);
   }
-
-  if (streaming_pad_probe_type(&t->s_streaming_pad, SMT_PACKET)) {
-    parse_mpeg_ts(t, st, buf, len, 1, 0);
-    if (t->s_hbbtv) {
-      if (!htsmsg_binary_serialize(t->s_hbbtv, &bin, &binlen, 128*1024)) {
-        parse_mpeg_ts(t, st, bin, binlen, 1, 0);
-        free(bin);
-      }
-    }
-  }
-
-dvberr:
-  htsmsg_destroy(apps);
-  htsmsg_destroy(titles);
-  return;
-}
\ No newline at end of file
+}
diff --git a/src/input/mpegts/tsdemux.c b/src/input/mpegts/tsdemux.c
index 36a9a9959..beef0fcc5 100644
--- a/src/input/mpegts/tsdemux.c
+++ b/src/input/mpegts/tsdemux.c
@@ -29,110 +29,6 @@
 static void ts_remux(mpegts_service_t *t, const uint8_t *tsb, int len, int errors);
 static void ts_skip(mpegts_service_t *t, const uint8_t *tsb, int len);
 
-/**
- * Extract PCR clocks
- */
-static void ts_recv_pcr(mpegts_service_t *t, const uint8_t *tsb)
-{
-  int64_t pcr;
-  pcr  =  (uint64_t)tsb[6] << 25;
-  pcr |=  (uint64_t)tsb[7] << 17;
-  pcr |=  (uint64_t)tsb[8] << 9;
-  pcr |=  (uint64_t)tsb[9] << 1;
-  pcr |= ((uint64_t)tsb[10] >> 7) & 0x01;
-  /* handle the broken info using candidate variable */
-  if (t->s_current_pcr == PTS_UNSET || t->s_current_pcr_guess ||
-      pts_diff(t->s_current_pcr, pcr) <= (int64_t)t->s_pcr_boundary ||
-      (t->s_candidate_pcr != PTS_UNSET &&
-       pts_diff(t->s_candidate_pcr, pcr) <= (int64_t)t->s_pcr_boundary)) {
-    if (pcr != t->s_current_pcr) {
-      if (t->s_current_pcr == PTS_UNSET)
-        tvhtrace(LS_PCR, "%s: initial  : %"PRId64, service_nicename((service_t*)t), pcr);
-      else
-        tvhtrace(LS_PCR, "%s: change   : %"PRId64"%s", service_nicename((service_t*)t), pcr,
-                         t->s_candidate_pcr != PTS_UNSET ? " (candidate)" : "");
-      t->s_current_pcr = pcr;
-      t->s_current_pcr_guess = 0;
-    }
-    t->s_candidate_pcr = PTS_UNSET;
-  } else {
-    tvhtrace(LS_PCR, "%s: candidate: %"PRId64, service_nicename((service_t*)t), pcr);
-    t->s_candidate_pcr = pcr;
-  }
-}
-
-static void ts_recv_pcr_audio
-  (mpegts_service_t *t, elementary_stream_t *st, const uint8_t *buf, int len)
-{
-  int64_t dts, pts, d;
-  int hdr, flags, hlen;
-
-  if (buf[3] != 0xbd && buf[3] != 0xc0)
-    return;
-
-  if (len < 9)
-    return;
-
-  buf  += 6;
-  len  -= 6;
-  hdr   = buf[0];
-  flags = buf[1];
-  hlen  = buf[2];
-  buf  += 3;
-  len  -= 3;
-
-  if (len < hlen || (hdr & 0xc0) != 0x80)
-    return;
-
-  if ((flags & 0xc0) == 0xc0) {
-    if (hlen < 10)
-      return;
-
-    pts = getpts(buf);
-    dts = getpts(buf + 5);
-
-    d = (pts - dts) & PTS_MASK;
-    if (d > 180000) // More than two seconds of PTS/DTS delta, probably corrupt
-      return;
-
-  } else if ((flags & 0xc0) == 0x80) {
-    if (hlen < 5)
-      return;
-
-    dts = pts = getpts(buf);
-  } else
-    return;
-
-  if (st->es_last_pcr == PTS_UNSET) {
-    d = pts_diff(st->es_last_pcr_dts, dts);
-    goto set;
-  }
-
-  if (st->es_last_pcr != t->s_current_pcr) {
-    st->es_last_pcr = t->s_current_pcr;
-    st->es_last_pcr_dts = dts;
-    return;
-  }
-
-  d = pts_diff(st->es_last_pcr_dts, dts);
-  if (d == PTS_UNSET || d < 30000)
-    return;
-  if (d < 10*90000)
-    t->s_current_pcr += d;
-
-set:
-  if (t->s_current_pcr == PTS_UNSET && d != PTS_UNSET && d < 30000) {
-    t->s_current_pcr = (dts - 2*90000) & PTS_MASK;
-    t->s_current_pcr_guess = 1;
-  }
-
-  tvhtrace(LS_PCR, "%s: audio DTS: %"PRId64" dts %"PRId64" [%s/%d]",
-                   service_nicename((service_t*)t), t->s_current_pcr, dts,
-                   streaming_component_type2txt(st->es_type), st->es_pid);
-  st->es_last_pcr = t->s_current_pcr;
-  st->es_last_pcr_dts = dts;
-}
-
 /**
  * Continue processing of transport stream packets
  */
@@ -141,7 +37,7 @@ ts_recv_packet0
   (mpegts_service_t *t, elementary_stream_t *st, const uint8_t *tsb, int len)
 {
   mpegts_service_t *m;
-  int len2, off, pusi, cc, pid, error, errors = 0;
+  int len2, off, cc, pid, error, errors = 0;
   const uint8_t *tsb2;
 
   service_set_streaming_status_flags((service_t*)t, TSS_MUX_PACKETS);
@@ -151,7 +47,6 @@ ts_recv_packet0
 
   for (tsb2 = tsb, len2 = len; len2 > 0; tsb2 += 188, len2 -= 188) {
 
-    pusi    = (tsb2[1] >> 6) & 1; /* 0x40 */
     error   = (tsb2[1] >> 7) & 1; /* 0x80 */
     errors += error;
 
@@ -175,34 +70,10 @@ ts_recv_packet0
     if (tsb2[3] & 0xc0) /* scrambled */
       continue;
 
-    if (tsb2[3] & 0x20) {
-      off = tsb2[4] + 5;
-      if (st->es_pid == t->s_pcr_pid && !error && off > 10 &&
-          (tsb2[5] & 0x10) != 0 && off <= 188)
-        ts_recv_pcr(t, tsb2);
-    } else {
-      off = 4;
-    }
-
-    if (pusi && !error && off < 188 - 16 &&
-        tsb2[off] == 0 && tsb2[off+1] == 0 && tsb2[off+2] == 1 &&
-        SCT_ISAUDIO(st->es_type))
-      ts_recv_pcr_audio(t, st, tsb2 + off, 188 - off);
-
     if (st->es_type == SCT_HBBTV) {
       dvb_table_parse(&st->es_psi, "ts", tsb2, 188, 1, 0, ts_recv_hbbtv_cb);
       continue;
     }
-
-    if (!streaming_pad_probe_type(&t->s_streaming_pad, SMT_PACKET))
-      continue;
-
-    if (st->es_type == SCT_CA || st->es_type == SCT_CAT)
-      continue;
-
-    if (off <= 188 && t->s_status == SERVICE_RUNNING)
-      parse_mpeg_ts((service_t*)t, st, tsb2 + off, 188 - off, pusi, error);
-
   }
 
   if (!t->s_scrambled_pass && (st->es_type == SCT_CA || st->es_type == SCT_CAT))
@@ -259,10 +130,6 @@ ts_recv_skipped0
 
   }
 
-  if(st->es_type != SCT_CA &&
-     streaming_pad_probe_type(&t->s_streaming_pad, SMT_PACKET))
-    skip_mpeg_ts((service_t*)t, st, tsb, len);
-
 skip_cc:
   if(streaming_pad_probe_type(&t->s_streaming_pad, SMT_MPEGTS))
     ts_skip(t, tsb, len);
diff --git a/src/input/mpegts/tsdemux.h b/src/input/mpegts/tsdemux.h
index d90840d77..b6858f4fe 100644
--- a/src/input/mpegts/tsdemux.h
+++ b/src/input/mpegts/tsdemux.h
@@ -33,6 +33,9 @@ void ts_skip_packet2(struct mpegts_service *t, const uint8_t *tsb, int len);
 
 void ts_recv_raw(struct mpegts_service *t, const uint8_t *tsb, int len);
 
+htsmsg_t *ts_recv_hbbtv(struct mpegts_psi_table *mt, elementary_stream_t *st,
+                        const uint8_t *buf, int len, int *_sect);
+
 void ts_recv_hbbtv_cb(mpegts_psi_table_t *mt, const uint8_t *buf, int len);
 
 #endif /* TSDEMUX_H */
diff --git a/src/muxer/muxer_libav.c b/src/muxer/muxer_libav.c
index 15719b957..914d83e21 100644
--- a/src/muxer/muxer_libav.c
+++ b/src/muxer/muxer_libav.c
@@ -28,6 +28,7 @@
 #include "channels.h"
 #include "libav.h"
 #include "muxer_libav.h"
+#include "parsers.h"
 #include "parsers/parser_avc.h"
 #include "parsers/parser_hevc.h"
 
diff --git a/src/muxer/muxer_mkv.c b/src/muxer/muxer_mkv.c
index 10216ab60..7a8ad0179 100644
--- a/src/muxer/muxer_mkv.c
+++ b/src/muxer/muxer_mkv.c
@@ -38,6 +38,7 @@
 #include "ebml.h"
 #include "lang_codes.h"
 #include "epg.h"
+#include "parsers.h"
 #include "parsers/parser_avc.h"
 #include "parsers/parser_hevc.h"
 #include "muxer_mkv.h"
diff --git a/src/parsers.h b/src/parsers.h
index daf0701e0..54136f6cc 100644
--- a/src/parsers.h
+++ b/src/parsers.h
@@ -19,8 +19,83 @@
 #ifndef PARSERS_H
 #define PARSERS_H
 
+#include "tvheadend.h"
+#include "streaming.h"
+#include "subscriptions.h"
 #include "packet.h"
 
+typedef struct parser_es parser_es_t;
+typedef struct parser parser_t;
+
+struct th_subscription;
+
+/* parser elementary stream */
+struct parser_es {
+  elementary_stream_t;
+  /* Parent */
+  parser_t *es_parser;
+  /* State */
+  sbuf_t    es_buf;
+  uint8_t   es_incomplete;
+  uint8_t   es_header_mode;
+  uint32_t  es_header_offset;
+  uint32_t  es_startcond;
+  uint32_t  es_startcode;
+  uint32_t  es_startcode_offset;
+  int       es_parser_state;
+  int       es_parser_ptr;
+  int       es_meta_change;
+  void     *es_priv;          /* Parser private data */
+  sbuf_t    es_buf_a;         /* Audio packet reassembly */
+  uint8_t  *es_global_data;
+  int       es_global_data_len;
+  struct th_pkt *es_curpkt;
+  struct streaming_message_queue es_backlog;
+  tvhlog_limit_t es_pes_log;
+  /* Clocks */
+  int64_t   es_curpts;
+  int64_t   es_curdts;
+  int64_t   es_prevdts;
+  int64_t   es_nextdts;
+  /* Misc */
+  char      es_blank;         /* Teletext: last subtitle was blank */
+  /* PCR clocks */
+  int64_t   es_last_pcr;
+  int64_t   es_last_pcr_dts;
+  tvhlog_limit_t es_pcr_log;
+};
+
+/* parser internal structure */
+struct parser {
+
+  streaming_target_t prs_input;
+
+  streaming_target_t *prs_output;
+
+  th_subscription_t *prs_subscription;
+
+  service_t *prs_service;
+
+  /* Elementary streams */
+  int prs_es_size;
+  int prs_es_count;
+  parser_es_t *prs_es;
+
+  /* Globals */
+  uint16_t prs_pcr_pid;
+
+  /* PCR clocks */
+  int64_t  prs_current_pcr;
+  int64_t  prs_candidate_pcr;
+  int64_t  prs_pcr_boundary;
+  uint8_t  prs_current_pcr_guess;
+
+  /* Teletext */
+  th_commercial_advice_t prs_tt_commercial_advice;
+  time_t prs_tt_clock;   /* Network clock as determined by teletext decoder */
+
+};
+
 static inline int64_t
 getpts(const uint8_t *p)
 {
@@ -42,18 +117,16 @@ getpts(const uint8_t *p)
   }
 }
 
-void parse_mpeg_ts(struct service *t, struct elementary_stream *st,
-		   const uint8_t *data, 
-		   int len, int start, int err);
+streaming_target_t * parser_create(streaming_target_t *output, struct th_subscription *ts);
+
+void parser_destroy(streaming_target_t *pad);
 
-void skip_mpeg_ts(struct service *t, struct elementary_stream *st,
-                  const uint8_t *data, int len);
+void parse_mpeg_ts(parser_t *t, parser_es_t *st, const uint8_t *data,
+                   int len, int start, int err);
 
-void parser_enqueue_packet(struct service *t, struct elementary_stream *st,
-			   th_pkt_t *pkt);
+void parser_enqueue_packet(struct service *t, parser_es_t *st, th_pkt_t *pkt);
 
-void parser_set_stream_vparam(struct elementary_stream *st, int width, int height,
-                              int duration);
+void parser_set_stream_vparam(parser_es_t *st, int width, int height, int duration);
 
 extern const unsigned int mpeg2video_framedurations[16];
 
diff --git a/src/parsers/message.c b/src/parsers/message.c
new file mode 100644
index 000000000..3ac5b2c79
--- /dev/null
+++ b/src/parsers/message.c
@@ -0,0 +1,404 @@
+/*
+ *  Packet parsing functions - streaming message handler
+ *  Copyright (C) 2017 Jaroslav Kysela
+ *
+ *  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 <pthread.h>
+
+#include <stdio.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <string.h>
+#include <errno.h>
+#include <assert.h>
+
+#include "parsers.h"
+#include "../input/mpegts/tsdemux.h"
+#include "packet.h"
+#include "streaming.h"
+#include "config.h"
+#include "htsmsg_binary2.h"
+
+/**
+ *
+ */
+static parser_es_t *
+parser_find_es(parser_t *prs, int pid)
+{
+  parser_es_t *pes;
+  int i;
+
+  for (i = 0, pes = prs->prs_es; i < prs->prs_es_count; i++, pes++)
+    if (pes->es_pid == pid)
+      return pes;
+  return NULL;
+}
+
+/**
+ * Extract PCR clocks
+ */
+static void ts_recv_pcr(parser_t *t, const uint8_t *tsb)
+{
+  int64_t pcr;
+  pcr  =  (uint64_t)tsb[6] << 25;
+  pcr |=  (uint64_t)tsb[7] << 17;
+  pcr |=  (uint64_t)tsb[8] << 9;
+  pcr |=  (uint64_t)tsb[9] << 1;
+  pcr |= ((uint64_t)tsb[10] >> 7) & 0x01;
+  /* handle the broken info using candidate variable */
+  if (t->prs_current_pcr == PTS_UNSET || t->prs_current_pcr_guess ||
+      pts_diff(t->prs_current_pcr, pcr) <= (int64_t)t->prs_pcr_boundary ||
+      (t->prs_candidate_pcr != PTS_UNSET &&
+       pts_diff(t->prs_candidate_pcr, pcr) <= (int64_t)t->prs_pcr_boundary)) {
+    if (pcr != t->prs_current_pcr) {
+      if (t->prs_current_pcr == PTS_UNSET)
+        tvhtrace(LS_PCR, "%s: initial  : %"PRId64, service_nicename(t->prs_service), pcr);
+      else
+        tvhtrace(LS_PCR, "%s: change   : %"PRId64"%s", service_nicename(t->prs_service), pcr,
+                         t->prs_candidate_pcr != PTS_UNSET ? " (candidate)" : "");
+      t->prs_current_pcr = pcr;
+      t->prs_current_pcr_guess = 0;
+    }
+    t->prs_candidate_pcr = PTS_UNSET;
+  } else {
+    tvhtrace(LS_PCR, "%s: candidate: %"PRId64, service_nicename(t->prs_service), pcr);
+    t->prs_candidate_pcr = pcr;
+  }
+}
+
+/*
+ * Extract PCR from audio
+ */
+static void ts_recv_pcr_audio
+  (parser_t *t, parser_es_t *st, const uint8_t *buf, int len)
+{
+  int64_t dts, pts, d;
+  int hdr, flags, hlen;
+
+  if (buf[3] != 0xbd && buf[3] != 0xc0)
+    return;
+
+  if (len < 9)
+    return;
+
+  buf  += 6;
+  len  -= 6;
+  hdr   = buf[0];
+  flags = buf[1];
+  hlen  = buf[2];
+  buf  += 3;
+  len  -= 3;
+
+  if (len < hlen || (hdr & 0xc0) != 0x80)
+    return;
+
+  if ((flags & 0xc0) == 0xc0) {
+    if (hlen < 10)
+      return;
+
+    pts = getpts(buf);
+    dts = getpts(buf + 5);
+
+    d = (pts - dts) & PTS_MASK;
+    if (d > 180000) // More than two seconds of PTS/DTS delta, probably corrupt
+      return;
+
+  } else if ((flags & 0xc0) == 0x80) {
+    if (hlen < 5)
+      return;
+
+    dts = pts = getpts(buf);
+  } else
+    return;
+
+  if (st->es_last_pcr == PTS_UNSET) {
+    d = pts_diff(st->es_last_pcr_dts, dts);
+    goto set;
+  }
+
+  if (st->es_last_pcr != t->prs_current_pcr) {
+    st->es_last_pcr = t->prs_current_pcr;
+    st->es_last_pcr_dts = dts;
+    return;
+  }
+
+  d = pts_diff(st->es_last_pcr_dts, dts);
+  if (d == PTS_UNSET || d < 30000)
+    return;
+  if (d < 10*90000)
+    t->prs_current_pcr += d;
+
+set:
+  if (t->prs_current_pcr == PTS_UNSET && d != PTS_UNSET && d < 30000) {
+    t->prs_current_pcr = (dts - 2*90000) & PTS_MASK;
+    t->prs_current_pcr_guess = 1;
+  }
+
+  tvhtrace(LS_PCR, "%s: audio DTS: %"PRId64" dts %"PRId64" [%s/%d]",
+                   service_nicename(t->prs_service), t->prs_current_pcr, dts,
+                   streaming_component_type2txt(st->es_type), st->es_pid);
+  st->es_last_pcr = t->prs_current_pcr;
+  st->es_last_pcr_dts = dts;
+}
+
+/**
+ * HBBTV
+ */
+static void
+ts_recv_hbbtv1_cb(mpegts_psi_table_t *mt, const uint8_t *buf, int len)
+{
+  parser_es_t *pes = (parser_es_t *)mt->mt_opaque;
+  parser_t *prs = pes->es_parser;
+  htsmsg_t *apps = ts_recv_hbbtv(mt, (elementary_stream_t *)pes, buf, len, NULL);
+
+  if (apps) {
+    void *bin;
+    size_t binlen;
+    parse_mpeg_ts(prs, pes, buf, len, 1, 0);
+    if (!htsmsg_binary2_serialize(apps, &bin, &binlen, 128*1024)) {
+      parse_mpeg_ts(prs, pes, bin, binlen, 1, 0);
+      free(bin);
+    }
+  }
+}
+
+/**
+ *
+ */
+static void parser_input_mpegts(parser_t *prs, pktbuf_t *pb)
+{
+  const uint8_t *tsb = pb->pb_data;
+  size_t len = pb->pb_size;
+  int err = pb->pb_err;
+  int pid, last_pid = -1;
+  int pusi, error, off;
+  parser_es_t *pes = NULL;
+
+  for (; len > 0; tsb += 188, len -= 188) {
+    pid = (tsb[1] & 0x1f) << 8 | tsb[2];
+    if (pid != last_pid) {
+      pes = parser_find_es(prs, pid);
+      last_pid = pid;
+      if (err && pes) {
+        /* FIXME: we have global error count */
+        /*        assign errors more precisely */
+        sbuf_err(&pes->es_buf, err);
+        err = 0;
+      }
+    }
+    if (pes == NULL)
+      continue;
+
+    pusi  = (tsb[1] >> 6) & 1; /* 0x40 */
+    error = (tsb[1] >> 7) & 1; /* 0x80 */
+
+    if (tsb[3] & 0x20) {
+      off = tsb[4] + 5;
+      if (pes->es_pid == prs->prs_pcr_pid && !error && off > 10 &&
+          (tsb[5] & 0x10) != 0 && off <= 188)
+        ts_recv_pcr(prs, tsb);
+    } else {
+      off = 4;
+    }
+
+    if (pusi && !error && off < 188 - 16 &&
+        tsb[off] == 0 && tsb[off+1] == 0 && tsb[off+2] == 1 &&
+        SCT_ISAUDIO(pes->es_type)) {
+      ts_recv_pcr_audio(prs, pes, tsb + off, 188 - off);
+    } else if (pes->es_type == SCT_HBBTV) {
+      dvb_table_parse(&pes->es_psi, "parser", tsb, 188, 1, 0, ts_recv_hbbtv1_cb);
+      continue;
+    }
+
+    if (off <= 188)
+      parse_mpeg_ts(prs, pes, tsb + off, 188 - off, pusi, error);
+  }
+}
+
+/**
+ *
+ */
+static void parser_init_es(parser_es_t *pes, parser_t *prs,
+                           streaming_component_type_t type)
+{
+  pes->es_parser = prs;
+  pes->es_type = type;
+  pes->es_incomplete = 0;
+  pes->es_header_mode = 0;
+  pes->es_parser_state = 0;
+  pes->es_blank = 0;
+  pes->es_startcond = 0xffffffff;
+  pes->es_curdts = PTS_UNSET;
+  pes->es_curpts = PTS_UNSET;
+  pes->es_prevdts = PTS_UNSET;
+  pes->es_last_pcr = PTS_UNSET;
+  pes->es_last_pcr_dts = PTS_UNSET;
+  TAILQ_INIT(&pes->es_backlog);
+  if (pes->es_type == SCT_HBBTV && pes->es_psi.mt_name == NULL)
+    dvb_table_parse_init(&pes->es_psi, "hbbtv", LS_TS, pes->es_pid,
+                         DVB_HBBTV_BASE, DVB_HBBTV_MASK, pes);
+}
+
+/**
+ *
+ */
+static void parser_clean_es(parser_es_t *pes)
+{
+  free(pes->es_priv);
+  pes->es_priv = NULL;
+
+  streaming_queue_clear(&pes->es_backlog);
+
+  if (pes->es_psi.mt_name)
+    dvb_table_reset(&pes->es_psi);
+
+  pes->es_startcode = 0;
+
+  sbuf_free(&pes->es_buf);
+  sbuf_free(&pes->es_buf_a);
+
+  if(pes->es_curpkt != NULL) {
+    pkt_ref_dec(pes->es_curpkt);
+    pes->es_curpkt = NULL;
+  }
+
+  free(pes->es_global_data);
+  pes->es_global_data = NULL;
+  pes->es_global_data_len = 0;
+
+  tvhlog_limit_reset(&pes->es_pes_log);
+  tvhlog_limit_reset(&pes->es_pcr_log);
+
+  free(pes->es_nicename);
+  pes->es_nicename = NULL;
+}
+
+/**
+ *
+ */
+static void parser_input_start(parser_t *prs, streaming_message_t *sm)
+{
+  streaming_start_t *ss = sm->sm_data;
+  const streaming_start_component_t *ssc;
+  parser_es_t *pes;
+  char buf[256];
+  int i;
+
+  for (i = 0; i < prs->prs_es_count; i++)
+    parser_clean_es(&prs->prs_es[i]);
+
+  if (prs->prs_es_size < ss->ss_num_components) {
+    prs->prs_es = realloc(prs->prs_es, ss->ss_num_components * sizeof(parser_es_t));
+    if (prs->prs_es == NULL)
+      abort();
+  }
+
+  prs->prs_es_count = ss->ss_num_components;
+  for (i = 0; i < ss->ss_num_components; i++) {
+    pes = &prs->prs_es[i];
+    ssc = &ss->ss_components[i];
+    parser_init_es(&prs->prs_es[i], prs, ssc->ssc_type);
+    pes->es_service = prs->prs_service;
+    pes->es_index = ssc->ssc_index;
+    pes->es_pid = ssc->ssc_pid;
+    if (pes->es_pid != -1) {
+      snprintf(buf, sizeof(buf), "%s: %s @ #%d",
+               service_nicename(prs->prs_service),
+               streaming_component_type2txt(pes->es_type),
+               pes->es_pid);
+    } else {
+      snprintf(buf, sizeof(buf), "%s: %s",
+               service_nicename(prs->prs_service),
+               streaming_component_type2txt(pes->es_type));
+    }
+    pes->es_nicename = strdup(buf);
+  }
+
+  prs->prs_current_pcr = PTS_UNSET;
+  prs->prs_candidate_pcr = PTS_UNSET;
+  prs->prs_current_pcr_guess = 0;
+  prs->prs_pcr_boundary = 90000;
+  if (service_has_no_audio(prs->prs_service, 1))
+    prs->prs_pcr_boundary = 6*90000;
+
+  streaming_target_deliver2(prs->prs_output, sm);
+}
+
+/**
+ *
+ */
+static void
+parser_input(void *opaque, streaming_message_t *sm)
+{
+  parser_t *prs = opaque;
+
+  switch(sm->sm_type) {
+  case SMT_MPEGTS:
+    parser_input_mpegts(prs, (pktbuf_t *)sm->sm_data);
+    streaming_msg_free(sm);
+    break;
+  case SMT_START:
+    parser_input_start(prs, sm);
+    break;
+  default:
+    streaming_target_deliver2(prs->prs_output, sm);
+    break;
+  }
+}
+
+static htsmsg_t *
+parser_input_info(void *opaque, htsmsg_t *list)
+{
+  parser_t *prs = opaque;
+  streaming_target_t *st = prs->prs_output;
+  htsmsg_add_str(list, NULL, "parser input");
+  return st->st_ops.st_info(st->st_opaque, list);
+}
+
+static streaming_ops_t parser_input_ops = {
+  .st_cb   = parser_input,
+  .st_info = parser_input_info
+};
+
+/**
+ * Parser create
+ */
+streaming_target_t *
+parser_create(streaming_target_t *output, th_subscription_t *ts)
+{
+  parser_t *prs = calloc(1, sizeof(parser_t));
+
+  prs->prs_output = output;
+  prs->prs_subscription = ts;
+  prs->prs_service = ts->ths_service;
+  streaming_target_init(&prs->prs_input, &parser_input_ops, prs, 0);
+  return &prs->prs_input;
+
+}
+
+/*
+ * Parser destroy
+ */
+void
+parser_destroy(streaming_target_t *pad)
+{
+  parser_t *prs = (parser_t *)pad;
+  int i;
+
+  for (i = 0; i < prs->prs_es_count; i++)
+    parser_clean_es(&prs->prs_es[i]);
+  free(prs);
+}
diff --git a/src/parsers/parser_avc.c b/src/parsers/parser_avc.c
index 10ed33c3d..eb6d2d6a9 100644
--- a/src/parsers/parser_avc.c
+++ b/src/parsers/parser_avc.c
@@ -19,6 +19,7 @@
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  */
 
+#include "parsers.h"
 #include "parser_avc.h"
 #include "parser_h264.h"
 #include "bitstream.h"
diff --git a/src/parsers/parser_h264.c b/src/parsers/parser_h264.c
index 05fc6ae8e..ade299a63 100644
--- a/src/parsers/parser_h264.c
+++ b/src/parsers/parser_h264.c
@@ -220,7 +220,7 @@ decode_scaling_list(bitstream_t *bs, int size)
 
 
 int
-h264_decode_seq_parameter_set(elementary_stream_t *st, bitstream_t *bs)
+h264_decode_seq_parameter_set(parser_es_t *st, bitstream_t *bs)
 {
   uint32_t profile_idc, level_idc, poc_type;
   uint32_t sps_id, tmp, i, width, height;
@@ -341,7 +341,7 @@ h264_decode_seq_parameter_set(elementary_stream_t *st, bitstream_t *bs)
 
 
 int
-h264_decode_pic_parameter_set(elementary_stream_t *st, bitstream_t *bs)
+h264_decode_pic_parameter_set(parser_es_t *st, bitstream_t *bs)
 {
   h264_private_t *p;
   uint32_t pps_id, sps_id;
@@ -368,7 +368,7 @@ h264_decode_pic_parameter_set(elementary_stream_t *st, bitstream_t *bs)
 
 
 int
-h264_decode_slice_header(elementary_stream_t *st, bitstream_t *bs, int *pkttype,
+h264_decode_slice_header(parser_es_t *st, bitstream_t *bs, int *pkttype,
 			 int *isfield)
 {
   h264_private_t *p;
diff --git a/src/parsers/parser_h264.h b/src/parsers/parser_h264.h
index a2ae2b9ad..79d8241fe 100644
--- a/src/parsers/parser_h264.h
+++ b/src/parsers/parser_h264.h
@@ -38,11 +38,11 @@
 
 void *h264_nal_deescape(bitstream_t *bs, const uint8_t *data, int size);
 
-int h264_decode_seq_parameter_set(struct elementary_stream *st, bitstream_t *bs);
+int h264_decode_seq_parameter_set(parser_es_t *st, bitstream_t *bs);
 
-int h264_decode_pic_parameter_set(struct elementary_stream *st, bitstream_t *bs);
+int h264_decode_pic_parameter_set(parser_es_t *st, bitstream_t *bs);
 
-int h264_decode_slice_header(struct elementary_stream *st, bitstream_t *bs,
+int h264_decode_slice_header(parser_es_t *st, bitstream_t *bs,
 			     int *pkttype, int *isfield);
 
 #endif /* PARSER_H264_H_ */
diff --git a/src/parsers/parser_hevc.c b/src/parsers/parser_hevc.c
index 4354149a6..90e251a81 100644
--- a/src/parsers/parser_hevc.c
+++ b/src/parsers/parser_hevc.c
@@ -1285,7 +1285,7 @@ static inline int check_height(uint32_t h)
 }
 
 void
-hevc_decode_vps(elementary_stream_t *st, bitstream_t *bs)
+hevc_decode_vps(parser_es_t *st, bitstream_t *bs)
 {
   hevc_private_t *p;
   hevc_vps_t *vps;
@@ -1418,7 +1418,7 @@ hevc_decode_vui(hevc_vui_t *vui, bitstream_t *bs)
 }
 
 void
-hevc_decode_sps(elementary_stream_t *st, bitstream_t *bs)
+hevc_decode_sps(parser_es_t *st, bitstream_t *bs)
 {
   hevc_private_t *p;
   hevc_vps_t *vps;
@@ -1597,7 +1597,7 @@ hevc_decode_sps(elementary_stream_t *st, bitstream_t *bs)
 }
 
 void
-hevc_decode_pps(elementary_stream_t *st, bitstream_t *bs)
+hevc_decode_pps(parser_es_t *st, bitstream_t *bs)
 {
   hevc_private_t *p;
   hevc_pps_t *pps;
@@ -1629,8 +1629,7 @@ hevc_decode_pps(elementary_stream_t *st, bitstream_t *bs)
 }
 
 int
-hevc_decode_slice_header(struct elementary_stream *st, bitstream_t *bs,
-                         int *pkttype)
+hevc_decode_slice_header(parser_es_t *st, bitstream_t *bs, int *pkttype)
 {
   hevc_private_t *p;
   hevc_vps_t *vps;
diff --git a/src/parsers/parser_hevc.h b/src/parsers/parser_hevc.h
index 08a7d80b7..af56f8422 100644
--- a/src/parsers/parser_hevc.h
+++ b/src/parsers/parser_hevc.h
@@ -58,11 +58,10 @@ int isom_write_hvcc(sbuf_t *pb, const uint8_t *src, int size);
 
 th_pkt_t * hevc_convert_pkt(th_pkt_t *src);
 
-void hevc_decode_vps(struct elementary_stream *st, bitstream_t *bs);
-void hevc_decode_sps(struct elementary_stream *st, bitstream_t *bs);
-void hevc_decode_pps(struct elementary_stream *st, bitstream_t *bs);
+void hevc_decode_vps(parser_es_t *st, bitstream_t *bs);
+void hevc_decode_sps(parser_es_t *st, bitstream_t *bs);
+void hevc_decode_pps(parser_es_t *st, bitstream_t *bs);
 
-int hevc_decode_slice_header(struct elementary_stream *st, bitstream_t *bs,
-                             int *pkttype);
+int hevc_decode_slice_header(parser_es_t *st, bitstream_t *bs, int *pkttype);
 
 #endif /* PARSER_HEVC_H */
diff --git a/src/parsers/parser_latm.c b/src/parsers/parser_latm.c
index fbcde2664..5f21f1ddf 100644
--- a/src/parsers/parser_latm.c
+++ b/src/parsers/parser_latm.c
@@ -93,8 +93,7 @@ static inline int read_sr(bitstream_t *bs, int *sri)
 }
 
 static int
-read_audio_specific_config(elementary_stream_t *st, latm_private_t *latm,
-			   bitstream_t *bs)
+read_audio_specific_config(parser_es_t *st, latm_private_t *latm, bitstream_t *bs)
 {
   int aot, sr, sri;
 
@@ -144,7 +143,7 @@ read_audio_specific_config(elementary_stream_t *st, latm_private_t *latm,
 
 
 static int
-read_stream_mux_config(elementary_stream_t *st, latm_private_t *latm, bitstream_t *bs)
+read_stream_mux_config(parser_es_t *st, latm_private_t *latm, bitstream_t *bs)
 {
   int audio_mux_version = read_bits1(bs);
   latm->audio_mux_version_A = 0;
@@ -225,7 +224,7 @@ read_stream_mux_config(elementary_stream_t *st, latm_private_t *latm, bitstream_
  * Parse AAC LATM
  */
 th_pkt_t *
-parse_latm_audio_mux_element(service_t *t, elementary_stream_t *st, 
+parse_latm_audio_mux_element(parser_t *t, parser_es_t *st,
 			     const uint8_t *data, int len)
 {
   latm_private_t *latm;
@@ -260,9 +259,10 @@ parse_latm_audio_mux_element(service_t *t, elementary_stream_t *st,
   if(st->es_curdts == PTS_UNSET)
     return NULL;
 
-  th_pkt_t *pkt = pkt_alloc(st->es_type, NULL, slot_len + 7, st->es_curdts, st->es_curdts, t->s_current_pcr);
+  th_pkt_t *pkt = pkt_alloc(st->es_type, NULL, slot_len + 7,
+                            st->es_curdts, st->es_curdts, t->prs_current_pcr);
 
-  pkt->pkt_commercial = t->s_tt_commercial_advice;
+  pkt->pkt_commercial = t->prs_tt_commercial_advice;
   pkt->pkt_duration   = st->es_frame_duration;
   pkt->a.pkt_sri      = latm->sri;
   pkt->a.pkt_ext_sri  = latm->ext_sri;
diff --git a/src/parsers/parser_latm.h b/src/parsers/parser_latm.h
index 583f1ddbc..4251bb2f4 100644
--- a/src/parsers/parser_latm.h
+++ b/src/parsers/parser_latm.h
@@ -19,8 +19,8 @@
 #ifndef PARSER_LATM_H_
 #define PARSER_LATM_H_
 
-th_pkt_t *parse_latm_audio_mux_element(struct service *t,
-				       struct elementary_stream *st,
+th_pkt_t *parse_latm_audio_mux_element(parser_t *t,
+				       parser_es_t *st,
 				       const uint8_t *data, int len);
 
 #endif /* PARSER_LATM_H_ */
diff --git a/src/parsers/parser_teletext.c b/src/parsers/parser_teletext.c
index bfebb4066..c0bd51248 100644
--- a/src/parsers/parser_teletext.c
+++ b/src/parsers/parser_teletext.c
@@ -37,6 +37,7 @@
 #include "streaming.h"
 #include "service.h"
 #include "input.h"
+#include "parsers.h"
 #include "parser_teletext.h"
 
 /**
@@ -65,7 +66,7 @@ typedef struct tt_private {
 
 static void teletext_rundown_copy(tt_private_t *ttp, tt_mag_t *ttm);
 
-static void teletext_rundown_scan(mpegts_service_t *t, tt_private_t *ttp);
+static void teletext_rundown_scan(parser_t *t, tt_private_t *ttp);
 
 #define bitreverse(b) \
 (((b) * 0x0202020202ULL & 0x010884422010ULL) % 1023)
@@ -662,7 +663,7 @@ is_tt_clock(const uint8_t *str)
  *
  */
 static int
-update_tt_clock(mpegts_service_t *t, const uint8_t *buf)
+update_tt_clock(parser_t *t, const uint8_t *buf)
 {
   uint8_t str[10];
   int i;
@@ -676,10 +677,10 @@ update_tt_clock(mpegts_service_t *t, const uint8_t *buf)
     return 0;
 
   ti = tt_construct_unix_time(str);
-  if(t->s_tt_clock == ti)
+  if(t->prs_tt_clock == ti)
     return 0;
 
-  t->s_tt_clock = ti;
+  t->prs_tt_clock = ti;
   //  printf("teletext clock is: %s", ctime(&ti));
   return 1;
 }
@@ -695,8 +696,7 @@ get_cset(uint8_t off)
 }
 
 static int
-extract_subtitle(mpegts_service_t *t, elementary_stream_t *st,
-		 tt_mag_t *ttm, int64_t pts)
+extract_subtitle(parser_t *t, parser_es_t *st, tt_mag_t *ttm, int64_t pts)
 {
   int i, j, start, off = 0;
   int k, current_color, is_font_tag_open, use_color_subs = 1;
@@ -796,7 +796,7 @@ extract_subtitle(mpegts_service_t *t, elementary_stream_t *st,
   th_pkt_t *pkt = pkt_alloc(st->es_type, sub, off, pts, pts, pts);
   pkt->pkt_componentindex = st->es_index;
 
-  streaming_service_deliver((service_t *)t, streaming_msg_create_pkt(pkt));
+  streaming_target_deliver2(t->prs_output, streaming_msg_create_pkt(pkt));
 
   /* Decrease our own reference to the packet */
   pkt_ref_dec(pkt);
@@ -836,19 +836,21 @@ dump_page(tt_mag_t *ttm)
 
 
 static void
-tt_subtitle_deliver(mpegts_service_t *t, elementary_stream_t *parent, tt_mag_t *ttm)
+tt_subtitle_deliver(parser_t *t, parser_es_t *parent, tt_mag_t *ttm)
 {
-  elementary_stream_t *st;
+  int i;
+  parser_es_t *st;
 
   if(ttm->ttm_current_pts == PTS_UNSET)
     return;
 
-  TAILQ_FOREACH(st, &t->s_components, es_link) {
-     if(parent->es_pid == st->es_parent_pid &&
-	ttm->ttm_curpage == st->es_pid -  PID_TELETEXT_BASE) {
-       if (extract_subtitle(t, st, ttm, ttm->ttm_current_pts))
-         ttm->ttm_current_pts++; // Avoid duplicate (non-monotonic) PTS
-     }
+  for (i = 0; i < t->prs_es_count; i++) {
+    st = &t->prs_es[i];
+    if (parent->es_pid == st->es_parent_pid &&
+      ttm->ttm_curpage == st->es_pid -  PID_TELETEXT_BASE) {
+      if (extract_subtitle(t, st, ttm, ttm->ttm_current_pts))
+        ttm->ttm_current_pts++; // Avoid duplicate (non-monotonic) PTS
+    }
   }
 }
 
@@ -856,7 +858,7 @@ tt_subtitle_deliver(mpegts_service_t *t, elementary_stream_t *parent, tt_mag_t *
  *
  */
 static void
-tt_decode_line(mpegts_service_t *t, elementary_stream_t *st, uint8_t *buf)
+tt_decode_line(parser_t *t, parser_es_t *st, uint8_t *buf)
 {
   uint8_t mpag, line, s12, c;
   int page, magidx, i;
@@ -914,7 +916,7 @@ tt_decode_line(mpegts_service_t *t, elementary_stream_t *st, uint8_t *buf)
     if(update_tt_clock(t, buf + 34))
       teletext_rundown_scan(t, ttp);
 
-    ttm->ttm_current_pts = t->s_current_pcr + (int64_t)t->s_pts_shift * 900;
+    ttm->ttm_current_pts = t->prs_current_pcr + (int64_t)st->es_service->s_pts_shift * 900;
     break;
 
   case 1 ... 23:
@@ -945,7 +947,7 @@ tt_decode_line(mpegts_service_t *t, elementary_stream_t *st, uint8_t *buf)
  */
 void
 teletext_input
-  (mpegts_service_t *t, elementary_stream_t *st, const uint8_t *data, int len)
+  (parser_t *t, parser_es_t *st, const uint8_t *data, int len)
 {
   int j;
   uint8_t buf[42];
@@ -1028,16 +1030,24 @@ teletext_rundown_copy(tt_private_t *ttp, tt_mag_t *ttm)
 
 
 static void
-teletext_rundown_scan(mpegts_service_t *t, tt_private_t *ttp)
+teletext_rundown_scan(parser_t *prs, tt_private_t *ttp)
 {
   int i;
   uint8_t *l;
-  time_t now = t->s_tt_clock, start, stop;
+  mpegts_service_t *t;
+  time_t now = prs->prs_tt_clock, start, stop;
   th_commercial_advice_t ca;
 
   if(ttp->ttp_rundown_valid == 0)
     return;
 
+  if(prs->prs_es_count <= 0)
+    return;
+
+  t = (mpegts_service_t *)prs->prs_es[0].es_service;
+  if(t == NULL)
+    return;
+
   if(t->s_dvb_svcname &&
      strcmp("TV4", t->s_dvb_svcname) &&
      strcmp("TV4 HD", t->s_dvb_svcname))
@@ -1057,6 +1067,6 @@ teletext_rundown_scan(mpegts_service_t *t, tt_private_t *ttp)
     stop  = start + tt_time_to_len(l + 32);
     
     if(start <= now && stop > now)
-      t->s_tt_commercial_advice = ca;
+      prs->prs_tt_commercial_advice = ca;
   }
 }
diff --git a/src/parsers/parser_teletext.h b/src/parsers/parser_teletext.h
index daa941811..2f44b386d 100644
--- a/src/parsers/parser_teletext.h
+++ b/src/parsers/parser_teletext.h
@@ -21,9 +21,6 @@
 
 #define PID_TELETEXT_BASE 0x2000
 
-#include "input.h"
-
-void teletext_input(struct mpegts_service *t, struct elementary_stream *st,
-		    const uint8_t *data, int len);
+void teletext_input(parser_t *t, parser_es_t *st, const uint8_t *data, int len);
 
 #endif /* TELETEXT_H */
diff --git a/src/parsers/parsers.c b/src/parsers/parsers.c
index 511efe76e..4957862d0 100644
--- a/src/parsers/parsers.c
+++ b/src/parsers/parsers.c
@@ -1,6 +1,7 @@
 /*
  *  Packet parsing functions
  *  Copyright (C) 2007 Andreas Öman
+ *  Copyright (C) 2014-2017 Jaroslav Kysela
  *
  *  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
@@ -25,8 +26,6 @@
 #include <errno.h>
 #include <assert.h>
 
-#include "tvheadend.h"
-#include "service.h"
 #include "parsers.h"
 #include "parser_h264.h"
 #include "parser_avc.h"
@@ -66,56 +65,56 @@ pts_no_backlog(int64_t pts)
   return pts_is_backlog(pts) ? (pts & ~PTS_BACKLOG) : pts;
 }
 
-static int parse_mpeg2video(service_t *t, elementary_stream_t *st, size_t len,
+static int parse_mpeg2video(parser_t *t, parser_es_t *st, size_t len,
                             uint32_t next_startcode, int sc_offset);
 
-static int parse_h264(service_t *t, elementary_stream_t *st, size_t len,
+static int parse_h264(parser_t *t, parser_es_t *st, size_t len,
                       uint32_t next_startcode, int sc_offset);
 
-static int parse_hevc(service_t *t, elementary_stream_t *st, size_t len,
+static int parse_hevc(parser_t *t, parser_es_t *st, size_t len,
                       uint32_t next_startcode, int sc_offset);
 
-typedef int (packet_parser_t)(service_t *t, elementary_stream_t *st, size_t len,
+typedef int (packet_parser_t)(parser_t *t, parser_es_t *st, size_t len,
                               uint32_t next_startcode, int sc_offset);
 
-typedef void (aparser_t)(service_t *t, elementary_stream_t *st, th_pkt_t *pkt);
+typedef void (aparser_t)(parser_t *t, parser_es_t *st, th_pkt_t *pkt);
 
-static void parse_pes(service_t *t, elementary_stream_t *st, const uint8_t *data,
+static void parse_pes(parser_t *t, parser_es_t *st, const uint8_t *data,
                       int len, int start, packet_parser_t *vp);
 
-static void parse_mp4a_data(service_t *t, elementary_stream_t *st, int skip_next_check);
+static void parse_mp4a_data(parser_t *t, parser_es_t *st, int skip_next_check);
 
-static void parse_aac(service_t *t, elementary_stream_t *st, const uint8_t *data,
+static void parse_aac(parser_t *t, parser_es_t *st, const uint8_t *data,
                       int len, int start);
 
-static void parse_subtitles(service_t *t, elementary_stream_t *st, 
+static void parse_subtitles(parser_t *t, parser_es_t *st,
                             const uint8_t *data, int len, int start);
 
-static void parse_teletext(service_t *t, elementary_stream_t *st, 
+static void parse_teletext(parser_t *t, parser_es_t *st,
                            const uint8_t *data, int len, int start);
 
-static void parse_hbbtv(service_t *t, elementary_stream_t *st,
+static void parse_hbbtv(parser_t *t, parser_es_t *st,
                         const uint8_t *data, int len, int start);
 
-static int parse_mpa(service_t *t, elementary_stream_t *st, size_t len,
+static int parse_mpa(parser_t *t, parser_es_t *st, size_t len,
                      uint32_t next_startcode, int sc_offset);
 
-static int parse_mpa123(service_t *t, elementary_stream_t *st);
+static int parse_mpa123(parser_t *t, parser_es_t *st);
 
-static int parse_ac3(service_t *t, elementary_stream_t *st, size_t len,
+static int parse_ac3(parser_t *t, parser_es_t *st, size_t len,
                      uint32_t next_startcode, int sc_offset);
 
-static void parser_deliver(service_t *t, elementary_stream_t *st, th_pkt_t *pkt);
+static void parser_deliver(parser_t *t, parser_es_t *st, th_pkt_t *pkt);
 
-static void parser_deliver_error(service_t *t, elementary_stream_t *st);
+static void parser_deliver_error(parser_t *t, parser_es_t *st);
 
-static int parse_pes_header(service_t *t, elementary_stream_t *st,
+static int parse_pes_header(parser_t *t, parser_es_t *st,
                             const uint8_t *buf, size_t len);
 
-static void parser_backlog(service_t *t, elementary_stream_t *st, th_pkt_t *pkt);
+static void parser_backlog(parser_t *t, parser_es_t *st, th_pkt_t *pkt);
 
-static void parser_do_backlog(service_t *t, elementary_stream_t *st,
-                  void (*pkt_cb)(service_t *t, elementary_stream_t *st, th_pkt_t *pkt),
+static void parser_do_backlog(parser_t *t, parser_es_t *st,
+                  void (*pkt_cb)(parser_t *t, parser_es_t *st, th_pkt_t *pkt),
                   pktbuf_t *meta);
 
 /**
@@ -144,10 +143,10 @@ static inline int data_noise(const uint8_t *data, int len) { return 0; }
 #endif
 
 /**
- * Parse raw mpeg data
+ * Parse raw mpeg data - single TS packet
  */
 void
-parse_mpeg_ts(service_t *t, elementary_stream_t *st, const uint8_t *data, 
+parse_mpeg_ts(parser_t *t, parser_es_t *st, const uint8_t *data,
               int len, int start, int err)
 {
   if(err) {
@@ -203,25 +202,11 @@ parse_mpeg_ts(service_t *t, elementary_stream_t *st, const uint8_t *data,
   }
 }
 
-/**
- * Skip raw mpeg data
- */
-void
-skip_mpeg_ts(service_t *t, elementary_stream_t *st, const uint8_t *data, int len)
-{
-  if(len >= 188)
-    sbuf_err(&st->es_buf, len / 188);
-  if(st->es_buf.sb_err > 1000) {
-    parser_deliver_error(t, st);
-    service_send_streaming_status((service_t *)t);
-  }
-}
-
 /**
  * Parse AAC LATM and ADTS
  */
 static void 
-parse_aac(service_t *t, elementary_stream_t *st, const uint8_t *data,
+parse_aac(parser_t *t, parser_es_t *st, const uint8_t *data,
           int len, int start)
 {
   int l, muxlen, p, latm;
@@ -334,7 +319,7 @@ parse_aac(service_t *t, elementary_stream_t *st, const uint8_t *data,
  * derive further information.
  */
 static void
-parse_pes(service_t *t, elementary_stream_t *st, const uint8_t *data, int len,
+parse_pes(parser_t *t, parser_es_t *st, const uint8_t *data, int len,
           int start, packet_parser_t *vp)
 {
   uint_fast32_t sc = st->es_startcond;
@@ -346,8 +331,7 @@ parse_pes(service_t *t, elementary_stream_t *st, const uint8_t *data, int len,
     if (data[0] != 0 && data[1] != 0 && data[2] != 1)
       if (tvhlog_limit(&st->es_pes_log, 10))
         tvhwarn(LS_TS, "%s: Invalid start code %02x:%02x:%02x",
-                service_component_nicename(st),
-                data[0], data[1], data[2]);
+                st->es_nicename, data[0], data[1], data[2]);
     st->es_incomplete = 0;
     if (st->es_header_mode) {
       st->es_buf.sb_ptr = st->es_header_offset;
@@ -467,7 +451,7 @@ found:
  *
  */
 static int
-depacketize(service_t *t, elementary_stream_t *st, size_t len, 
+depacketize(parser_t *t, parser_es_t *st, size_t len,
             uint32_t next_startcode, int sc_offset)
 {
   const uint8_t *buf = st->es_buf.sb_data + sc_offset;
@@ -507,12 +491,12 @@ depacketize(service_t *t, elementary_stream_t *st, size_t len,
  *
  */
 static void
-makeapkt(service_t *t, elementary_stream_t *st, const void *buf,
+makeapkt(parser_t *t, parser_es_t *st, const void *buf,
          int len, int64_t dts, int duration, int channels, int sri)
 {
-  th_pkt_t *pkt = pkt_alloc(st->es_type, buf, len, dts, dts, t->s_current_pcr);
+  th_pkt_t *pkt = pkt_alloc(st->es_type, buf, len, dts, dts, t->prs_current_pcr);
 
-  pkt->pkt_commercial = t->s_tt_commercial_advice;
+  pkt->pkt_commercial = t->prs_tt_commercial_advice;
   pkt->pkt_duration = duration;
   pkt->a.pkt_channels = channels;
   pkt->a.pkt_sri = sri;
@@ -558,7 +542,7 @@ mp4a_valid_frame(const uint8_t *buf)
   return (buf[0] == 0xff) && ((buf[1] & 0xf6) == 0xf0);
 }
 
-static void parse_mp4a_data(service_t *t, elementary_stream_t *st,
+static void parse_mp4a_data(parser_t *t, parser_es_t *st,
                             int skip_next_check)
 {
   int i, len;
@@ -629,7 +613,7 @@ mpa_valid_frame(uint32_t h)
  *
  */
 static int
-parse_mpa123(service_t *t, elementary_stream_t *st)
+parse_mpa123(parser_t *t, parser_es_t *st)
 {
   int i, len, layer, lsf, mpeg25, br, sr, pad;
   int fsize, fsize2, channels, duration;
@@ -685,7 +669,7 @@ ok:
         tvhtrace(LS_PARSER, "mpeg audio version change %02d: val=%d (old=%d)",
                  st->es_index, layer, st->es_audio_version);
         st->es_audio_version = layer;
-        atomic_set(&t->s_pending_restart, 1);
+        atomic_set(&st->es_service->s_pending_restart, 1);
       }
       makeapkt(t, st, buf + i, fsize, dts, duration,
                channels, mpa_sri[(buf[i+2] >> 2) & 3]);
@@ -703,7 +687,7 @@ ok:
  *
  */
 static int 
-parse_mpa(service_t *t, elementary_stream_t *st, size_t ilen,
+parse_mpa(parser_t *t, parser_es_t *st, size_t ilen,
           uint32_t next_startcode, int sc_offset)
 {
   int r;
@@ -782,7 +766,7 @@ ac3_valid_frame(const uint8_t *buf)
 static const char acmodtab[8] = {2,1,2,3,3,4,4,5};
 
 static int 
-parse_ac3(service_t *t, elementary_stream_t *st, size_t ilen,
+parse_ac3(parser_t *t, parser_es_t *st, size_t ilen,
           uint32_t next_startcode, int sc_offset)
 {
   int i, len, count, ver, bsid, fscod, frmsizcod, fsize, fsize2, duration, sri;
@@ -901,7 +885,7 @@ ok:
  * Extract DTS and PTS and update current values in stream
  */
 static int
-parse_pes_header(service_t *t, elementary_stream_t *st, 
+parse_pes_header(parser_t *t, parser_es_t *st,
                  const uint8_t *buf, size_t len)
 {
   int64_t dts, pts, d, d2;
@@ -970,7 +954,7 @@ parse_pes_header(service_t *t, elementary_stream_t *st,
   st->es_curpts = PTS_UNSET;
   if (tvhlog_limit(&st->es_pes_log, 10))
     tvhwarn(LS_TS, "%s Corrupted PES header (errors %zi)",
-            service_component_nicename(st), st->es_pes_log.count);
+            st->es_nicename, st->es_pes_log.count);
   return -1;
 } 
 
@@ -999,7 +983,7 @@ const unsigned int mpeg2video_framedurations[16] = {
  * Parse mpeg2video picture start
  */
 static int
-parse_mpeg2video_pic_start(service_t *t, elementary_stream_t *st, int *frametype,
+parse_mpeg2video_pic_start(parser_t *t, parser_es_t *st, int *frametype,
                            bitstream_t *bs)
 {
   int pct;
@@ -1029,7 +1013,7 @@ parse_mpeg2video_pic_start(service_t *t, elementary_stream_t *st, int *frametype
  *
  */
 void
-parser_set_stream_vparam(elementary_stream_t *st, int width, int height,
+parser_set_stream_vparam(parser_es_t *st, int width, int height,
                          int duration)
 {
   int need_save = 0;
@@ -1086,7 +1070,7 @@ static const uint8_t mpeg2_aspect[16][2]={
  * Parse mpeg2video sequence start
  */
 static int
-parse_mpeg2video_seq_start(service_t *t, elementary_stream_t *st,
+parse_mpeg2video_seq_start(parser_t *t, parser_es_t *st,
                            bitstream_t *bs)
 {
   int width, height, aspect, duration;
@@ -1132,7 +1116,7 @@ drop_trailing_zeroes(const uint8_t *buf, size_t len)
  *
  */
 static void
-parser_global_data_move(elementary_stream_t *st, const uint8_t *data, size_t len)
+parser_global_data_move(parser_es_t *st, const uint8_t *data, size_t len)
 {
   int len2 = drop_trailing_zeroes(data, len);
 
@@ -1155,7 +1139,7 @@ parser_global_data_move(elementary_stream_t *st, const uint8_t *data, size_t len
  *
  */
 static int
-parse_mpeg2video(service_t *t, elementary_stream_t *st, size_t len, 
+parse_mpeg2video(parser_t *t, parser_es_t *st, size_t len,
                  uint32_t next_startcode, int sc_offset)
 {
   const uint8_t *buf = st->es_buf.sb_data + sc_offset;
@@ -1188,10 +1172,10 @@ parse_mpeg2video(service_t *t, elementary_stream_t *st, size_t len,
     if(st->es_curpkt != NULL)
       pkt_ref_dec(st->es_curpkt);
 
-    pkt = pkt_alloc(st->es_type, NULL, 0, st->es_curpts, st->es_curdts, t->s_current_pcr);
+    pkt = pkt_alloc(st->es_type, NULL, 0, st->es_curpts, st->es_curdts, t->prs_current_pcr);
     pkt->v.pkt_frametype = frametype;
     pkt->pkt_duration = st->es_frame_duration;
-    pkt->pkt_commercial = t->s_tt_commercial_advice;
+    pkt->pkt_commercial = t->prs_tt_commercial_advice;
 
     st->es_curpkt = pkt;
 
@@ -1305,7 +1289,7 @@ parse_mpeg2video(service_t *t, elementary_stream_t *st, size_t len,
  *
  */
 static void
-parse_h264_backlog(service_t *t, elementary_stream_t *st, th_pkt_t *pkt)
+parse_h264_backlog(parser_t *t, parser_es_t *st, th_pkt_t *pkt)
 {
   int len, l2, pkttype = 0, isfield = 0, nal_type;
   const uint8_t *end, *nal_start, *nal_end;
@@ -1346,7 +1330,7 @@ parse_h264_backlog(service_t *t, elementary_stream_t *st, th_pkt_t *pkt)
 }
 
 static void
-parse_h264_deliver(service_t *t, elementary_stream_t *st, th_pkt_t *pkt)
+parse_h264_deliver(parser_t *t, parser_es_t *st, th_pkt_t *pkt)
 {
   if (pkt->v.pkt_frametype > 0) {
     if (TAILQ_EMPTY(&st->es_backlog)) {
@@ -1370,7 +1354,7 @@ deliver:
 }
 
 static int
-parse_h264(service_t *t, elementary_stream_t *st, size_t len, 
+parse_h264(parser_t *t, parser_es_t *st, size_t len,
            uint32_t next_startcode, int sc_offset)
 {
   const uint8_t *buf = st->es_buf.sb_data + sc_offset;
@@ -1479,11 +1463,11 @@ parse_h264(service_t *t, elementary_stream_t *st, size_t len,
       if (st->es_frame_duration == 0)
         st->es_frame_duration = 1;
 
-      pkt = pkt_alloc(st->es_type, NULL, 0, st->es_curpts, st->es_curdts, t->s_current_pcr);
+      pkt = pkt_alloc(st->es_type, NULL, 0, st->es_curpts, st->es_curdts, t->prs_current_pcr);
       pkt->v.pkt_frametype = pkttype;
       pkt->v.pkt_field = isfield;
       pkt->pkt_duration = st->es_frame_duration;
-      pkt->pkt_commercial = t->s_tt_commercial_advice;
+      pkt->pkt_commercial = t->prs_tt_commercial_advice;
       st->es_curpkt = pkt;
       break;
 
@@ -1533,7 +1517,7 @@ parse_h264(service_t *t, elementary_stream_t *st, size_t len,
  *
  */
 static int
-parse_hevc(service_t *t, elementary_stream_t *st, size_t len,
+parse_hevc(parser_t *t, parser_es_t *st, size_t len,
            uint32_t next_startcode, int sc_offset)
 {
   const uint8_t *buf = st->es_buf.sb_data + sc_offset;
@@ -1604,11 +1588,11 @@ parse_hevc(service_t *t, elementary_stream_t *st, size_t len,
     if (r > 0)
       return PARSER_APPEND;
 
-    st->es_curpkt = pkt_alloc(st->es_type, NULL, 0, st->es_curpts, st->es_curdts, t->s_current_pcr);
+    st->es_curpkt = pkt_alloc(st->es_type, NULL, 0, st->es_curpts, st->es_curdts, t->prs_current_pcr);
     st->es_curpkt->v.pkt_frametype = pkttype;
     st->es_curpkt->v.pkt_field = 0;
     st->es_curpkt->pkt_duration = st->es_frame_duration;
-    st->es_curpkt->pkt_commercial = t->s_tt_commercial_advice;
+    st->es_curpkt->pkt_commercial = t->prs_tt_commercial_advice;
     break;
 
   case HEVC_NAL_VPS:
@@ -1696,7 +1680,7 @@ parse_hevc(service_t *t, elementary_stream_t *st, size_t len,
  * http://broadcasting.ru/pdf-standard-specifications/subtitling/dvb-sub/en300743.v1.2.1.pdf
  */
 static void
-parse_subtitles(service_t *t, elementary_stream_t *st, const uint8_t *data,
+parse_subtitles(parser_t *t, parser_es_t *st, const uint8_t *data,
                 int len, int start)
 {
   th_pkt_t *pkt;
@@ -1749,8 +1733,8 @@ parse_subtitles(service_t *t, elementary_stream_t *st, const uint8_t *data,
 
     // end_of_PES_data_field_marker
     if(buf[psize - 1] == 0xff) {
-      pkt = pkt_alloc(st->es_type, buf, psize - 1, st->es_curpts, st->es_curdts, t->s_current_pcr);
-      pkt->pkt_commercial = t->s_tt_commercial_advice;
+      pkt = pkt_alloc(st->es_type, buf, psize - 1, st->es_curpts, st->es_curdts, t->prs_current_pcr);
+      pkt->pkt_commercial = t->prs_tt_commercial_advice;
       pkt->pkt_err = st->es_buf.sb_err;
       parser_deliver(t, st, pkt);
       sbuf_reset(&st->es_buf, 4000);
@@ -1762,7 +1746,7 @@ parse_subtitles(service_t *t, elementary_stream_t *st, const uint8_t *data,
  * Teletext parser
  */
 static void
-parse_teletext(service_t *t, elementary_stream_t *st, const uint8_t *data,
+parse_teletext(parser_t *t, parser_es_t *st, const uint8_t *data,
                int len, int start)
 {
   th_pkt_t *pkt;
@@ -1799,10 +1783,11 @@ parse_teletext(service_t *t, elementary_stream_t *st, const uint8_t *data,
   psize -= hlen;
   buf = d + 6 + hlen;
   
-  if(psize >= 46 && t->s_current_pcr != PTS_UNSET) {
-    teletext_input((mpegts_service_t *)t, st, buf, psize);
-    pkt = pkt_alloc(st->es_type, buf, psize, t->s_current_pcr, t->s_current_pcr, t->s_current_pcr);
-    pkt->pkt_commercial = t->s_tt_commercial_advice;
+  if(psize >= 46 && t->prs_current_pcr != PTS_UNSET) {
+    teletext_input(t, st, buf, psize);
+    pkt = pkt_alloc(st->es_type, buf, psize,
+                    t->prs_current_pcr, t->prs_current_pcr, t->prs_current_pcr);
+    pkt->pkt_commercial = t->prs_tt_commercial_advice;
     pkt->pkt_err = st->es_buf.sb_err;
     parser_deliver(t, st, pkt);
     sbuf_reset(&st->es_buf, 4000);
@@ -1813,10 +1798,11 @@ parse_teletext(service_t *t, elementary_stream_t *st, const uint8_t *data,
  * Hbbtv parser (=forwarder)
  */
 static void
-parse_hbbtv(service_t *t, elementary_stream_t *st, const uint8_t *data,
+parse_hbbtv(parser_t *t, parser_es_t *st, const uint8_t *data,
             int len, int start)
 {
-  th_pkt_t *pkt = pkt_alloc(st->es_type, data, len, t->s_current_pcr, t->s_current_pcr, t->s_current_pcr);
+  th_pkt_t *pkt = pkt_alloc(st->es_type, data, len,
+                            t->prs_current_pcr, t->prs_current_pcr, t->prs_current_pcr);
   pkt->pkt_err = st->es_buf.sb_err;
   parser_deliver(t, st, pkt);
 }
@@ -1825,11 +1811,12 @@ parse_hbbtv(service_t *t, elementary_stream_t *st, const uint8_t *data,
  *
  */
 static th_pkt_t *
-parser_error_packet(service_t *t, elementary_stream_t *st, int error)
+parser_error_packet(parser_t *t, parser_es_t *st, int error)
 {
   th_pkt_t *pkt;
 
-  pkt = pkt_alloc(st->es_type, NULL, 0, PTS_UNSET, PTS_UNSET, t->s_current_pcr);
+  pkt = pkt_alloc(st->es_type, NULL, 0,
+                  PTS_UNSET, PTS_UNSET, t->prs_current_pcr);
   pkt->pkt_err = error;
   return pkt;
 }
@@ -1838,7 +1825,7 @@ parser_error_packet(service_t *t, elementary_stream_t *st, int error)
  *
  */
 static void
-parser_deliver(service_t *t, elementary_stream_t *st, th_pkt_t *pkt)
+parser_deliver(parser_t *t, parser_es_t *st, th_pkt_t *pkt)
 {
   int64_t d, diff;
 
@@ -1859,7 +1846,7 @@ parser_deliver(service_t *t, elementary_stream_t *st, th_pkt_t *pkt)
   if (d > diff || d == PTS_UNSET) {
     if (d != PTS_UNSET && tvhlog_limit(&st->es_pcr_log, 2))
       tvhwarn(LS_PARSER, "%s: DTS and PCR diff is very big (%"PRId64")",
-              service_component_nicename(st), pts_diff(pkt->pkt_pcr, pkt->pkt_pts));
+              st->es_nicename, pts_diff(pkt->pkt_pcr, pkt->pkt_pts));
     pkt_trace(LS_PARSER, pkt, "pcr drop");
     pkt_ref_dec(pkt);
     pkt = parser_error_packet(t, st, 1);
@@ -1876,16 +1863,8 @@ deliver:
     pkt->v.pkt_aspect_den = st->es_aspect_den;
   }
 
-  /**
-   * Input is ok
-   */
-  if (pkt->pkt_payload) {
-    service_set_streaming_status_flags(t, TSS_PACKETS);
-    t->s_streaming_live |= TSS_LIVE;
-  }
-
   /* Forward packet */
-  streaming_service_deliver(t, streaming_msg_create_pkt(pkt));
+  streaming_target_deliver2(t->prs_output, streaming_msg_create_pkt(pkt));
 
   /* Decrease our own reference to the packet */
   pkt_ref_dec(pkt);
@@ -1896,7 +1875,7 @@ deliver:
  * Deliver errors
  */
 static void
-parser_deliver_error(service_t *t, elementary_stream_t *st)
+parser_deliver_error(parser_t *t, parser_es_t *st)
 {
   th_pkt_t *pkt;
 
@@ -1911,7 +1890,7 @@ parser_deliver_error(service_t *t, elementary_stream_t *st)
  * Do backlog (fix DTS & PTS and reparse slices - frame type, field etc.)
  */
 static void
-parser_backlog(service_t *t, elementary_stream_t *st, th_pkt_t *pkt)
+parser_backlog(parser_t *t, parser_es_t *st, th_pkt_t *pkt)
 {
   streaming_message_t *sm = streaming_msg_create_pkt(pkt);
   TAILQ_INSERT_TAIL(&st->es_backlog, sm, sm_link);
@@ -1931,8 +1910,8 @@ parser_backlog(service_t *t, elementary_stream_t *st, th_pkt_t *pkt)
 }
 
 static void
-parser_do_backlog(service_t *t, elementary_stream_t *st,
-                  void (*pkt_cb)(service_t *t, elementary_stream_t *st, th_pkt_t *pkt),
+parser_do_backlog(parser_t *t, parser_es_t *st,
+                  void (*pkt_cb)(parser_t *t, parser_es_t *st, th_pkt_t *pkt),
                   pktbuf_t *meta)
 {
   streaming_message_t *sm;
diff --git a/src/service.c b/src/service.c
index f5b2f63ee..9ff6705a3 100644
--- a/src/service.c
+++ b/src/service.c
@@ -271,24 +271,9 @@ stream_init(elementary_stream_t *st)
 {
   st->es_cc = -1;
 
-  st->es_last_pcr = PTS_UNSET;
-  st->es_last_pcr_dts = PTS_UNSET;
-
-  st->es_incomplete = 0;
-  st->es_header_mode = 0;
-  st->es_parser_state = 0;
-  st->es_startcond = 0xffffffff;
-  st->es_curdts = PTS_UNSET;
-  st->es_curpts = PTS_UNSET;
-  st->es_prevdts = PTS_UNSET;
-
-  st->es_blank = 0;
-
   if (st->es_type == SCT_HBBTV && st->es_psi.mt_name == NULL)
     dvb_table_parse_init(&st->es_psi, "hbbtv", LS_TS, st->es_pid,
                          DVB_HBBTV_BASE, DVB_HBBTV_MASK, st);
-
-  TAILQ_INIT(&st->es_backlog);
 }
 
 
@@ -298,33 +283,7 @@ stream_init(elementary_stream_t *st)
 static void
 stream_clean(elementary_stream_t *st)
 {
-  free(st->es_priv);
-  st->es_priv = NULL;
-
-  /* Clear reassembly buffers */
-
-  streaming_queue_clear(&st->es_backlog);
-
-  st->es_startcode = 0;
-
-  sbuf_free(&st->es_buf);
-  sbuf_free(&st->es_buf_a);
-
-  if(st->es_curpkt != NULL) {
-    pkt_ref_dec(st->es_curpkt);
-    st->es_curpkt = NULL;
-  }
-
-  free(st->es_global_data);
-  st->es_global_data = NULL;
-  st->es_global_data_len = 0;
-
-  free(st->es_section);
-  st->es_section = NULL;
-
   tvhlog_limit_reset(&st->es_cc_log);
-  tvhlog_limit_reset(&st->es_pes_log);
-  tvhlog_limit_reset(&st->es_pcr_log);
 
   if (st->es_psi.mt_name)
     dvb_table_reset(&st->es_psi);
@@ -362,7 +321,6 @@ service_stream_destroy(service_t *t, elementary_stream_t *es)
     free(c);
   }
 
-  free(es->es_section);
   free(es->es_nicename);
   free(es);
 }
@@ -707,12 +665,9 @@ service_start(service_t *t, int instance, int weight, int flags,
   t->s_scrambled_seen   = 0;
   t->s_scrambled_pass   = !!(flags & SUBSCRIPTION_NODESCR);
   t->s_start_time       = mclk();
-  t->s_pcr_boundary     = 90000;
 
   pthread_mutex_lock(&t->s_stream_mutex);
   service_build_filter(t);
-  if (service_has_no_audio(t, 1))
-    t->s_pcr_boundary = 6*90000;
   pthread_mutex_unlock(&t->s_stream_mutex);
 
   descrambler_caid_changed(t);
@@ -725,9 +680,6 @@ service_start(service_t *t, int instance, int weight, int flags,
   pthread_mutex_lock(&t->s_stream_mutex);
 
   t->s_status = SERVICE_RUNNING;
-  t->s_current_pcr = PTS_UNSET;
-  t->s_candidate_pcr = PTS_UNSET;
-  t->s_current_pcr_guess = 0;
 
   /**
    * Initialize stream
diff --git a/src/service.h b/src/service.h
index e6c1e6ae3..f9602734b 100644
--- a/src/service.h
+++ b/src/service.h
@@ -61,88 +61,48 @@ typedef struct elementary_stream {
 
   TAILQ_ENTRY(elementary_stream) es_link;
   TAILQ_ENTRY(elementary_stream) es_filt_link;
-  int es_position;
+
+  uint32_t es_position;
   struct service *es_service;
 
   streaming_component_type_t es_type;
   int es_index;
 
-  uint16_t es_aspect_num;
-  uint16_t es_aspect_den;
-
-  char es_lang[4];           /* ISO 639 2B 3-letter language code */
-  uint8_t es_audio_type;     /* Audio type */
-  uint8_t es_audio_version;  /* Audio version/layer */
-
-  uint16_t es_composition_id;
-  uint16_t es_ancillary_id;
+  char *es_nicename;
 
+  /* PID related */
   int16_t es_pid;
   uint16_t es_parent_pid;    /* For subtitle streams originating from 
 				a teletext stream. this is the pid
 				of the teletext stream */
   int8_t es_pid_opened;      /* PID is opened */
+  int8_t es_cc;              /* Last CC */
 
-  int8_t es_cc;             /* Last CC */
-
-  int es_peak_presentation_delay; /* Max seen diff. of DTS and PTS */
-
-  /* PCR clocks */
-  int64_t  es_last_pcr;
-  int64_t  es_last_pcr_dts;
-
-  /* For service stream packet reassembly */
-
-  sbuf_t es_buf;
-
-  uint8_t  es_incomplete;
-  uint8_t  es_header_mode;
-  uint32_t es_header_offset;
-  uint32_t es_startcond;
-  uint32_t es_startcode;
-  uint32_t es_startcode_offset;
-  int es_parser_state;
-  int es_parser_ptr;
-  void *es_priv;          /* Parser private data */
-
-  sbuf_t es_buf_a;        // Audio packet reassembly
+  /* CA ID's on this stream */
+  struct caid_list es_caids;
 
-  uint8_t *es_global_data;
-  int es_global_data_len;
+  /* */
+  int es_delete_me;      /* Temporary flag for deleting streams */
 
-  struct th_pkt *es_curpkt;
-  struct streaming_message_queue es_backlog;
-  int64_t es_curpts;
-  int64_t es_curdts;
-  int64_t es_prevdts;
-  int64_t es_nextdts;
+  /* Stream info */
   int es_frame_duration;
+
   int es_width;
   int es_height;
 
-  int es_meta_change;
-
-  /* CA ID's on this stream */
-  struct caid_list es_caids;
+  uint16_t es_aspect_num;
+  uint16_t es_aspect_den;
 
-  /* */
+  char es_lang[4];           /* ISO 639 2B 3-letter language code */
+  uint8_t es_audio_type;     /* Audio type */
+  uint8_t es_audio_version;  /* Audio version/layer */
 
-  int es_delete_me;      /* Temporary flag for deleting streams */
+  uint16_t es_composition_id;
+  uint16_t es_ancillary_id;
 
   /* Error log limiters */
-
   tvhlog_limit_t es_cc_log;
-  tvhlog_limit_t es_pes_log;
-  tvhlog_limit_t es_pcr_log;
   
-  char *es_nicename;
-
-  /* Teletext subtitle */ 
-  char es_blank; // Last subtitle was blank
-
-  /* SI section processing (horrible hack) */
-  void *es_section;
-
   /* Filter temporary variable */
   uint32_t es_filter;
 
@@ -318,7 +278,6 @@ typedef struct service {
   int s_prio;
   int s_type_user;
   int s_pts_shift; // in ms (may be negative)
-  int s_pcr_boundary;
 
   LIST_ENTRY(service) s_active_link;
 
@@ -499,10 +458,6 @@ typedef struct service {
 
   tvhlog_limit_t s_tei_log;
 
-  int64_t s_current_pcr;
-  int64_t s_candidate_pcr;
-  uint8_t s_current_pcr_guess;
-
   /*
    * Local channel numbers per bouquet
    */
diff --git a/src/subscriptions.c b/src/subscriptions.c
index 0b30b2af1..3f6e62323 100644
--- a/src/subscriptions.c
+++ b/src/subscriptions.c
@@ -34,6 +34,7 @@
 #include "tvheadend.h"
 #include "subscriptions.h"
 #include "streaming.h"
+#include "parsers.h"
 #include "channels.h"
 #include "service.h"
 #include "profile.h"
@@ -740,6 +741,11 @@ subscription_unsubscribe(th_subscription_t *s, int flags)
   mtimer_disarm(&s->ths_remove_timer);
   mtimer_disarm(&s->ths_ca_check_timer);
 
+  if (s->ths_parser) {
+    parser_destroy(s->ths_parser);
+    s->ths_parser = NULL;
+  }
+
   if ((flags & UNSUBSCRIBE_FINAL) != 0 ||
       (s->ths_flags & SUBSCRIPTION_ONESHOT) != 0)
     subscription_destroy(s);
@@ -764,32 +770,19 @@ subscription_create
   th_subscription_t *s = calloc(1, sizeof(th_subscription_t));
   profile_t *pro = prch ? prch->prch_pro : NULL;
   streaming_target_t *st = prch ? prch->prch_st : NULL;
-  int reject = 0;
   static int tally;
 
   TAILQ_INIT(&s->ths_instances);
 
-  switch (flags & SUBSCRIPTION_TYPE_MASK) {
-  case SUBSCRIPTION_NONE:
-    reject |= SMT_TO_MASK(SMT_PACKET) | SMT_TO_MASK(SMT_MPEGTS);
-    break;
-  case SUBSCRIPTION_MPEGTS:
-    reject |= SMT_TO_MASK(SMT_PACKET);  // Reject parsed frames
-    break;
-  case SUBSCRIPTION_PACKET:
-    reject |= SMT_TO_MASK(SMT_MPEGTS);  // Reject raw mpegts
-    break;
-  default:
-    abort();
-  }
-
   if (!ops) ops = &subscription_input_direct_ops;
   if (!st) {
     st = calloc(1, sizeof(streaming_target_t));
     streaming_target_init(st, &subscription_input_null_ops, s, 0);
+  } else if ((flags & SUBSCRIPTION_TYPE_MASK) == SUBSCRIPTION_PACKET) {
+    st = s->ths_parser = parser_create(st, s);
   }
 
-  streaming_target_init(&s->ths_input, ops, s, reject);
+  streaming_target_init(&s->ths_input, ops, s, 0);
 
   s->ths_prch              = prch && prch->prch_st ? prch : NULL;
   s->ths_title             = strdup(name);
diff --git a/src/subscriptions.h b/src/subscriptions.h
index aec5989d3..3ebc62a3d 100644
--- a/src/subscriptions.h
+++ b/src/subscriptions.h
@@ -109,6 +109,7 @@ typedef struct th_subscription {
   streaming_target_t ths_input;
 
   streaming_target_t *ths_output;
+  streaming_target_t *ths_parser;
 
   int ths_flags;
   int ths_timeout;
diff --git a/src/transcoding/transcode/helpers.c b/src/transcoding/transcode/helpers.c
index 8e6ef714d..918c863b0 100644
--- a/src/transcoding/transcode/helpers.c
+++ b/src/transcoding/transcode/helpers.c
@@ -20,6 +20,7 @@
 
 #include "internals.h"
 
+#include "parsers.h"
 #include "parsers/bitstream.h"
 #include "parsers/parser_avc.h"
 #include "parsers/parser_h264.h"