static gboolean
check_bin (KmsTreeBin * tree_bin, const GstCaps * caps)
{
  gboolean ret = FALSE;
  GstElement *output_tee = kms_tree_bin_get_output_tee (tree_bin);
  GstPad *tee_sink = gst_element_get_static_pad (output_tee, "sink");
  GstCaps *current_caps = kms_tree_bin_get_input_caps (tree_bin);

  if (current_caps == NULL) {
    current_caps = gst_pad_get_allowed_caps (tee_sink);
    GST_TRACE_OBJECT (tree_bin, "Allowed caps are: %" GST_PTR_FORMAT,
        current_caps);
  } else {
    GST_TRACE_OBJECT (tree_bin, "Current caps are: %" GST_PTR_FORMAT,
        current_caps);
  }

  if (current_caps != NULL) {
    //TODO: Remove this when problem in negotiation with features will be
    //resolved
    GstCaps *caps_without_features = gst_caps_make_writable (current_caps);

    gst_caps_set_features (caps_without_features, 0,
        gst_caps_features_new_empty ());
    if (gst_caps_can_intersect (caps, caps_without_features)) {
      ret = TRUE;
    }
    gst_caps_unref (caps_without_features);
  }

  g_object_unref (tee_sink);

  return ret;
}

gint
gst_dtls_connection_send (GstDtlsConnection * self, gpointer data, gint len)
{
  int ret = 0;

  g_return_val_if_fail (GST_IS_DTLS_CONNECTION (self), 0);

  g_return_val_if_fail (self->priv->ssl, 0);
  g_return_val_if_fail (self->priv->bio, 0);

  GST_TRACE_OBJECT (self, "locking @ send");
  g_mutex_lock (&self->priv->mutex);
  GST_TRACE_OBJECT (self, "locked @ send");

  if (SSL_is_init_finished (self->priv->ssl)) {
    ret = SSL_write (self->priv->ssl, data, len);
    GST_DEBUG_OBJECT (self, "data sent: input was %d B, output is %d B", len,
        ret);
  } else {
    GST_WARNING_OBJECT (self,
        "tried to send data before handshake was complete");
    ret = 0;
  }

  GST_TRACE_OBJECT (self, "unlocking @ send");
  g_mutex_unlock (&self->priv->mutex);

  return ret;
}

/* Convert to caps that can be accepted by this element... */
static GstCaps *
get_converted_caps (GstGLStereoMix * mix, GstCaps * caps)
{
#if 0
  GstGLContext *context = GST_GL_BASE_MIXER (mix)->context;
  GstCaps *result, *tmp;

  GST_LOG_OBJECT (mix, "Converting caps %" GST_PTR_FORMAT, caps);
  result = gst_gl_upload_transform_caps (context, GST_PAD_SINK, caps, NULL);
  tmp = result;
  GST_TRACE_OBJECT (mix, "transfer returned caps %" GST_PTR_FORMAT, tmp);

  result =
      gst_gl_color_convert_transform_caps (context, GST_PAD_SINK, tmp, NULL);
  gst_caps_unref (tmp);
  GST_TRACE_OBJECT (mix, "convert returned caps %" GST_PTR_FORMAT, tmp);

  tmp = result;
  result = gst_gl_view_convert_transform_caps (mix->viewconvert,
      GST_PAD_SINK, tmp, NULL);
  gst_caps_unref (tmp);
#else
  GstCaps *result;

  GST_LOG_OBJECT (mix, "Converting caps %" GST_PTR_FORMAT, caps);
  result = gst_gl_view_convert_transform_caps (mix->viewconvert,
      GST_PAD_SINK, caps, NULL);
#endif

  GST_LOG_OBJECT (mix, "returning caps %" GST_PTR_FORMAT, result);

  return result;
}

static gboolean
check_bin (KmsTreeBin * tree_bin, const GstCaps * caps)
{
  gboolean ret = FALSE;
  GstElement *output_tee = kms_tree_bin_get_output_tee (tree_bin);
  GstPad *tee_sink = gst_element_get_static_pad (output_tee, "sink");
  GstCaps *current_caps = kms_tree_bin_get_input_caps (tree_bin);

  if (current_caps == NULL) {
    current_caps = gst_pad_get_allowed_caps (tee_sink);
    GST_TRACE_OBJECT (tree_bin, "Allowed caps are: %" GST_PTR_FORMAT,
        current_caps);
  } else {
    gst_caps_ref (current_caps);
    GST_TRACE_OBJECT (tree_bin, "Current caps are: %" GST_PTR_FORMAT,
        current_caps);
  }

  if (current_caps != NULL) {
    if (gst_caps_can_intersect (caps, current_caps)) {
      ret = TRUE;
    }
    gst_caps_unref (current_caps);
  }

  g_object_unref (tee_sink);

  return ret;
}

static GstPadProbeReturn
kms_agnostic_bin2_sink_caps_probe (GstPad * pad, GstPadProbeInfo * info,
    gpointer user_data)
{
  KmsAgnosticBin2 *self;
  GstCaps *current_caps;
  GstCaps *new_caps = NULL;
  GstEvent *event = gst_pad_probe_info_get_event (info);

  if (GST_EVENT_TYPE (event) != GST_EVENT_CAPS) {
    return GST_PAD_PROBE_OK;
  }

  GST_TRACE_OBJECT (pad, "Event: %" GST_PTR_FORMAT, event);

  self = KMS_AGNOSTIC_BIN2 (user_data);

  gst_event_parse_caps (event, &new_caps);

  if (new_caps == NULL) {
    GST_ERROR_OBJECT (self, "Unexpected NULL caps");
    return GST_PAD_PROBE_OK;
  }

  KMS_AGNOSTIC_BIN2_LOCK (self);
  current_caps = self->priv->input_caps;
  self->priv->input_caps = gst_caps_copy (new_caps);
  KMS_AGNOSTIC_BIN2_UNLOCK (self);

  GST_TRACE_OBJECT (user_data, "New caps event: %" GST_PTR_FORMAT, event);

  if (current_caps != NULL) {
    GstStructure *st;

    GST_TRACE_OBJECT (user_data, "Current caps: %" GST_PTR_FORMAT,
        current_caps);

    st = gst_caps_get_structure (current_caps, 0);
    // Remove famerate, width, height, streamheader that make unecessary
    // agnostic reconstruction happen

    gst_structure_remove_fields (st, "width", "height", "framerate",
        "streamheader", "codec_data", NULL);

    if (!gst_caps_can_intersect (new_caps, current_caps) &&
        !is_raw_caps (current_caps) && !is_raw_caps (new_caps)) {
      GST_DEBUG_OBJECT (user_data, "Caps differ caps: %" GST_PTR_FORMAT,
          new_caps);
      kms_agnostic_bin2_configure_input (self, new_caps);
    }

    gst_caps_unref (current_caps);
  } else {
    GST_DEBUG_OBJECT (user_data, "No previous caps, starting");
    kms_agnostic_bin2_configure_input (self, new_caps);
  }

  return GST_PAD_PROBE_OK;
}

/* chain function
 * this function does the actual processing
 */
static GstFlowReturn gst_throttle_chain(GstPad * pad, GstBuffer * buf)
{
	GstThrottle * filter = GST_THROTTLE(GST_OBJECT_PARENT(pad));
	
	if (filter->printOnly)
	{
		GstCaps * caps = gst_buffer_get_caps(buf);
		gchar * capsStr = gst_caps_to_string(caps);
		gst_caps_unref(caps);
		GST_LOG_OBJECT(filter, "ts: %" GST_TIME_FORMAT " %sof type %s",
			GST_TIME_ARGS(buf->timestamp),
			GST_BUFFER_IS_DISCONT(buf) ? "and discontinuity " : "",
			capsStr
		);
		g_free(capsStr);
		
		GstFlowReturn ret = gst_pad_push(filter->srcpad, buf);
		GST_TRACE_OBJECT(filter, "ts: %" GST_TIME_FORMAT " processed with status %d", GST_TIME_ARGS(buf->timestamp), ret);
		return ret;
	}
	
	if (filter->clock == NULL)
	{
		return gst_pad_push(filter->srcpad, buf);
	}
	
	GstClockTime realTs = gst_clock_get_time(filter->clock);
	
	if (filter->haveStartTime)
	{
		const char * discont = GST_BUFFER_IS_DISCONT(buf) ? " with discotinuity" : "";
		
		GstClockTime expectedRealTs = filter->streamStartRealTime + buf->timestamp;
		gboolean early = realTs < expectedRealTs;
		if (early)
		{
			GstClockID * cid = gst_clock_new_single_shot_id(filter->clock, expectedRealTs);
			GST_TRACE_OBJECT(filter, "ts: %" GST_TIME_FORMAT " %s, waiting for %ld ms", GST_TIME_ARGS(buf->timestamp), discont, (expectedRealTs - realTs)/1000000);
			gst_clock_id_wait(cid, NULL);
			gst_clock_id_unref(cid);
		}
		else
		{
			GST_TRACE_OBJECT(filter, "ts: %" GST_TIME_FORMAT " %s, pad on time", GST_TIME_ARGS(buf->timestamp), discont);
		}
	}
	else
	{
		filter->streamStartRealTime = realTs - buf->timestamp;
		filter->haveStartTime = TRUE;
	}
	
	return gst_pad_push(filter->srcpad, buf);
}

gboolean
gst_gl_display_x11_handle_event (GstGLDisplayX11 * display_x11)
{
  xcb_connection_t *connection = display_x11->xcb_connection;
  xcb_generic_event_t *event;
  gboolean ret = TRUE;

  while ((event = xcb_poll_for_event (connection))) {
    GstGLWindowX11 *window_x11 = _window_from_event (display_x11, event);

    GST_TRACE_OBJECT (display_x11, "got event %p to window %" GST_PTR_FORMAT,
        event, window_x11);

    if (window_x11) {
      ret = gst_gl_window_x11_handle_event (window_x11, event);
    } else {
      /* unknown window, ignore */
      ret = TRUE;
    }

    if (window_x11)
      gst_object_unref (window_x11);
    g_free (event);
  }

  return ret;
}

void
RecorderEndpointImpl::onStateChanged (gint newState)
{
  switch (newState) {
  case KMS_URI_END_POINT_STATE_STOP: {
    GST_DEBUG_OBJECT (element, "State changed to Stopped");
    Stopped event (shared_from_this(), Stopped::getName() );
    signalStopped (event);
    break;
  }

  case KMS_URI_END_POINT_STATE_START: {
    GST_DEBUG_OBJECT (element, "State changed to Recording");
    Recording event (shared_from_this(), Recording::getName() );
    signalRecording (event);
    break;
  }

  case KMS_URI_END_POINT_STATE_PAUSE: {
    GST_DEBUG_OBJECT (element, "State changed to Paused");
    Paused event (shared_from_this(), Paused::getName() );
    signalPaused (event);
    break;
  }
  }

  std::unique_lock<std::mutex> lck (mtx);

  GST_TRACE_OBJECT (element, "State changed to %d", newState);

  state = newState;
  cv.notify_one();
}

static void
send_remb_event (KmsRembRemote * rm, guint bitrate, guint ssrc)
{
  GstEvent *event;
  guint br, min = 0, max = 0;

  /* TODO: use g_atomic */
  if (rm->pad_event == NULL) {
    return;
  }

  br = bitrate;

  if (rm->min_bw > 0) {
    min = rm->min_bw * 1000;
    br = MAX (br, min);
  }

  if (rm->max_bw > 0) {
    max = rm->max_bw * 1000;
    br = MIN (br, max);
  }

  GST_TRACE_OBJECT (KMS_REMB_BASE (rm)->rtpsess,
      "bitrate: %" G_GUINT32_FORMAT ", ssrc: %" G_GUINT32_FORMAT
      ", range [%" G_GUINT32_FORMAT ", %" G_GUINT32_FORMAT
      "], event bitrate: %" G_GUINT32_FORMAT, bitrate, ssrc, min, max, br);

  event = kms_utils_remb_event_upstream_new (br, ssrc);
  gst_pad_push_event (rm->pad_event, event);
}

static void
registry_handle_global (void *data, struct wl_registry *registry,
    uint32_t name, const char *interface, uint32_t version)
{
  GstGLWindowWaylandEGL *window_egl = data;
  struct display *d = &window_egl->display;

  GST_TRACE_OBJECT (window_egl, "registry_handle_global with registry %p, "
      "interface %s, version %u", registry, interface, version);

  if (g_strcmp0 (interface, "wl_compositor") == 0) {
    d->compositor =
        wl_registry_bind (registry, name, &wl_compositor_interface, 1);
  } else if (g_strcmp0 (interface, "wl_shell") == 0) {
    d->shell = wl_registry_bind (registry, name, &wl_shell_interface, 1);
  } else if (g_strcmp0 (interface, "wl_seat") == 0) {
    d->seat = wl_registry_bind (registry, name, &wl_seat_interface, 1);
    wl_seat_add_listener (d->seat, &seat_listener, window_egl);
  } else if (g_strcmp0 (interface, "wl_shm") == 0) {
    d->shm = wl_registry_bind (registry, name, &wl_shm_interface, 1);
    d->cursor_theme = wl_cursor_theme_load (NULL, 32, d->shm);
    d->default_cursor =
        wl_cursor_theme_get_cursor (d->cursor_theme, "left_ptr");
  }
}

static void
kms_ice_nice_agent_new_candidate (NiceAgent * agent,
    guint stream_id,
    guint component_id, gchar * foundation, KmsIceNiceAgent * self)
{
  KmsIceBaseAgent *parent = KMS_ICE_BASE_AGENT (self);
  GSList *candidates;
  GSList *walk;

  GST_TRACE_OBJECT (self,
      "stream_id: %d, component_id: %d, foundation: %s", stream_id,
      component_id, foundation);

  candidates = nice_agent_get_local_candidates (agent, stream_id, component_id);

  for (walk = candidates; walk; walk = walk->next) {
    NiceCandidate *cand = walk->data;

    if (cand->stream_id == stream_id &&
        cand->component_id == component_id &&
        g_strcmp0 (foundation, cand->foundation) == 0) {
      kms_ice_nice_agent_sdp_msg_add_ice_candidate (self->priv->session,
          self->priv->agent, cand, parent);
    }
  }
  g_slist_free_full (candidates, (GDestroyNotify) nice_candidate_free);
}

/**
 * owr_message_origin_post_message:
 * @origin: (transfer none): the origin that is posting the message
 * @type: the #OwrMessageType of the message
 * @sub_type: the #OwrMessageSubType of the message
 * @data: (element-type utf8 GValue) (nullable) (transfer full): extra data
 *
 * Post a new message to all buses that are subscribed to @origin
 */
void owr_message_origin_post_message(OwrMessageOrigin *origin, OwrMessageType type, OwrMessageSubType sub_type, GHashTable *data)
{
    OwrMessage *message;
    OwrMessageOriginBusSet *bus_set;
    GHashTableIter iter;
    GWeakRef *ref;
    OwrBus *bus;

    g_return_if_fail(OWR_IS_MESSAGE_ORIGIN(origin));

    message = _owr_message_new(origin, type, sub_type, data);
    GST_TRACE_OBJECT(origin, "posting message %p", message);

    bus_set = owr_message_origin_get_bus_set(origin);
    g_return_if_fail(bus_set);

    g_mutex_lock(&bus_set->mutex);

    g_hash_table_iter_init(&iter, bus_set->table);
    while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &ref)) {
        bus = g_weak_ref_get(ref);
        if (bus) {
            _owr_bus_post_message(bus, message);
            g_object_unref(bus);
        } else {
            GST_DEBUG_OBJECT(origin, "message bus finalized, removing weak ref: %p", ref);
            g_hash_table_iter_remove(&iter);
        }
    }

    _owr_message_unref(message);

    g_mutex_unlock(&bus_set->mutex);
}

void
gst_dtls_connection_check_timeout (GstDtlsConnection * self)
{
  GstDtlsConnectionPrivate *priv;

  g_return_if_fail (GST_IS_DTLS_CONNECTION (self));

  priv = self->priv;

  GST_TRACE_OBJECT (self, "locking @ start_timeout");
  g_mutex_lock (&priv->mutex);
  GST_TRACE_OBJECT (self, "locked @ start_timeout");
  gst_dtls_connection_check_timeout_locked (self);
  g_mutex_unlock (&priv->mutex);
  GST_TRACE_OBJECT (self, "unlocking @ start_timeout");
}

static gboolean
aggregate_func (GstAggregator * self)
{
  GstAggregatorPrivate *priv = self->priv;
  GstAggregatorClass *klass = GST_AGGREGATOR_GET_CLASS (self);

  GST_LOG_OBJECT (self, "Checking aggregate");
  while (priv->send_eos && gst_aggregator_iterate_sinkpads (self,
          (GstAggregatorPadForeachFunc) _check_all_pads_with_data_or_eos,
          NULL) && priv->running) {
    GST_TRACE_OBJECT (self, "Actually aggregating!");

    priv->flow_return = klass->aggregate (self);

    if (priv->flow_return == GST_FLOW_EOS) {
      g_main_context_wakeup (self->priv->mcontext);
      _remove_all_sources (self);
      _push_eos (self);
    }

    if (priv->flow_return == GST_FLOW_FLUSHING &&
        g_atomic_int_get (&priv->flush_seeking))
      priv->flow_return = GST_FLOW_OK;

    GST_LOG_OBJECT (self, "flow return is %s",
        gst_flow_get_name (priv->flow_return));

    if (priv->flow_return != GST_FLOW_OK)
      break;
  }

  return G_SOURCE_REMOVE;
}

static void
remb_event_manager_update_min (RembEventManager * manager, guint bitrate,
    guint ssrc)
{
  RembHashValue *last_value, *value;

  g_mutex_lock (&manager->mutex);
  last_value = g_hash_table_lookup (manager->remb_hash,
      GUINT_TO_POINTER (ssrc));

  if (last_value != NULL && bitrate == last_value->bitrate) {
    last_value->ts = kms_utils_get_time_nsecs ();
    goto end;
  }

  value = remb_hash_value_create (bitrate);
  g_hash_table_insert (manager->remb_hash, GUINT_TO_POINTER (ssrc), value);

  if (bitrate > manager->remb_min) {
    remb_event_manager_calc_min (manager);
  } else {
    remb_event_manager_set_min (manager, bitrate);
  }

end:

  GST_TRACE_OBJECT (manager->pad, "remb_min: %" G_GUINT32_FORMAT,
      manager->remb_min);

  g_mutex_unlock (&manager->mutex);
}

static GstPadProbeReturn
drop_until_keyframe_probe (GstPad * pad, GstPadProbeInfo * info,
    gpointer user_data)
{
  GstBuffer *buffer;
  gboolean all_headers = GPOINTER_TO_INT (user_data);

  buffer = GST_PAD_PROBE_INFO_BUFFER (info);

  if (GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT)) {
    /* Drop buffer until a keyframe is received */
    send_force_key_unit_event (pad, all_headers);
    GST_TRACE_OBJECT (pad, "Dropping buffer");
    return GST_PAD_PROBE_DROP;
  }

  GST_OBJECT_LOCK (pad);
  set_dropping (pad, FALSE);
  GST_OBJECT_UNLOCK (pad);

  GST_DEBUG_OBJECT (pad, "Finish dropping buffers until key frame");

  /* So this buffer is a keyframe we don't need this probe any more */
  return GST_PAD_PROBE_REMOVE;
}

static void
gst_openjpeg_dec_opj_warning (const char *msg, void *userdata)
{
  GstOpenJPEGDec *self = GST_OPENJPEG_DEC (userdata);
  gchar *trimmed = g_strchomp (g_strdup (msg));
  GST_TRACE_OBJECT (self, "openjpeg warning: %s", trimmed);
  g_free (trimmed);
}

static void
gst_openjpeg_enc_opj_info (const char *msg, void *userdata)
{
  GstOpenJPEGEnc *self = GST_OPENJPEG_ENC (userdata);
  gchar *trimmed = g_strchomp (g_strdup (msg));
  GST_TRACE_OBJECT (self, "openjpeg info: %s", trimmed);
  g_free (trimmed);
}

static void
_gst_gl_context_thread_run_generic (RunGenericData * data)
{
  GST_TRACE_OBJECT (data->context, "running function:%p data:%p", data->func,
      data->data);

  data->func (data->context, data->data);
}

static GstPadProbeReturn
control_duplicates (GstPad * pad, GstPadProbeInfo * info, gpointer data)
{
  GstEvent *event = GST_PAD_PROBE_INFO_EVENT (info);

  if (gst_video_event_is_force_key_unit (event)) {
    if (check_last_request_time (pad)) {
      GST_TRACE_OBJECT (pad, "Sending keyframe request");
      return GST_PAD_PROBE_OK;
    } else {
      GST_TRACE_OBJECT (pad, "Dropping keyframe request");
      return GST_PAD_PROBE_DROP;
    }
  }

  return GST_PAD_PROBE_OK;
}