text:
#include <gst/gst.h>
#include "MediaType.hpp"
#include "MediaLatencyStat.hpp"
#include "MediaType.hpp"
#include "AudioCaps.hpp"
#include "VideoCaps.hpp"
#include "AudioCodec.hpp"
#include "VideoCodec.hpp"
#include "Fraction.hpp"
#include "MediaElementImpl.hpp"
#include <jsonrpc/JsonSerializer.hpp>
#include <KurentoException.hpp>
#include <MediaPipelineImpl.hpp>
#include <MediaSet.hpp>
#include <gst/gst.h>
#include <ElementConnectionData.hpp>
#include <DotGraph.hpp>
#include <GstreamerDotDetails.hpp>
#include <StatsType.hpp>
#include "ElementStats.hpp"
#include "kmsstats.h"
#include <SignalHandler.hpp>

#define GST_CAT_DEFAULT kurento_media_element_impl
GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);
#define GST_DEFAULT_NAME "KurentoMediaElementImpl"

#define KMS_DEFAULT_MEDIA_DESCRIPTION "default"

#define MIN_OUTPUT_BITRATE "min-output-bitrate"
#define MAX_OUTPUT_BITRATE "max-output-bitrate"

#define TYPE_VIDEO "video_"
#define TYPE_AUDIO "audio_"

namespace kurento
{

const static std::string DEFAULT = "default";

class ElementConnectionDataInternal
{
public:
  ElementConnectionDataInternal (std::shared_ptr<MediaElement> source,
                                 std::shared_ptr<MediaElement> sink,
                                 std::shared_ptr<MediaType> type,
                                 const std::string &sourceDescription,
                                 const std::string &sinkDescription)
  {
    this->source = source;
    this->sink = sink;
    this->type = type;
    this->sourceDescription = sourceDescription;
    this->sinkDescription = sinkDescription;
    this->sourcePadName = NULL;
    setSinkPadName ();
  }

  ~ElementConnectionDataInternal()
  {
    if (sourcePadName != NULL) {
      free (sourcePadName);
    }
  }

  ElementConnectionDataInternal (std::shared_ptr<ElementConnectionData> data)
  {
    this->source = data->getSource();
    this->sink = data->getSink();
    this->type = data->getType();
    this->sourceDescription = data->getSourceDescription();
    this->sinkDescription = data->getSinkDescription();
    this->sourcePadName = NULL;
    setSinkPadName ();
  }

  void setSinkPadName()
  {
    std::string desc = "_" + (sinkDescription.empty () ?
                              KMS_DEFAULT_MEDIA_DESCRIPTION : sinkDescription);

    switch (type->getValue () ) {
    case MediaType::AUDIO:
      sinkPadName = "sink_audio" + desc;
      break;

    case MediaType::VIDEO:
      sinkPadName = "sink_video" + desc;
      break;

    case MediaType::DATA:
      sinkPadName = "sink_data" + desc;
      break;
    }
  }

  void setSourcePadName (gchar *padName)
  {
    if (this->sourcePadName != NULL) {
      GST_WARNING ("Resetting padName for connection");

      if (this->sourcePadName != padName) {
        free (this->sourcePadName);
      }
    }

    this->sourcePadName = padName;
  }

  const gchar *getSourcePadName ()
  {
    return sourcePadName;
  }

  std::shared_ptr<MediaElementImpl> getSource ()
  {
    try {
      return std::dynamic_pointer_cast <MediaElementImpl> (source.lock() );
    } catch (std::bad_cast) {
      GST_WARNING ("Bad cast for source element");
      return std::shared_ptr<MediaElementImpl> ();
    }
  }

  std::shared_ptr<MediaElementImpl> getSink ()
  {
    try {
      return std::dynamic_pointer_cast <MediaElementImpl> (sink.lock() );
    } catch (std::bad_cast) {
      GST_WARNING ("Bad cast for source element");
      return std::shared_ptr<MediaElementImpl> ();
    }
  }

  std::string getSinkPadName ()
  {
    return sinkPadName;
  }

  GstPad *getSinkPad ()
  {
    std::shared_ptr <MediaElementImpl> sinkLocked = getSink ();

    if (!sinkLocked) {
      return NULL;
    }

    return gst_element_get_static_pad (sinkLocked->getGstreamerElement (),
                                       getSinkPadName ().c_str() );
  }

  GstPad *getSourcePad ()
  {
    std::shared_ptr <MediaElementImpl> sourceLocked = getSource ();

    if (!sourceLocked || sourcePadName == NULL) {
      return NULL;
    }

    return gst_element_get_static_pad (sourceLocked->getGstreamerElement (),
                                       sourcePadName);
  }

  std::shared_ptr<ElementConnectionData> toInterface ()
  {
    std::shared_ptr<ElementConnectionData> iface (new ElementConnectionData (
          source.lock(), sink.lock(), type, sourceDescription, sinkDescription) );

    if (!iface->getSink () ) {
      throw KurentoException (MEDIA_OBJECT_NOT_FOUND, "Reference to sink is null");
    }

    if (!iface->getSource () ) {
      throw KurentoException (MEDIA_OBJECT_NOT_FOUND, "Reference to source is null");
    }

    return iface;
  }

private:

  std::weak_ptr<MediaElement> source;
  std::weak_ptr<MediaElement> sink;
  std::shared_ptr<MediaType> type;
  std::string sourceDescription;
  std::string sinkDescription;
  std::string sinkPadName;
  gchar *sourcePadName;
};

static KmsElementPadType
convertMediaType (std::shared_ptr<MediaType> mediaType)
{
  switch (mediaType->getValue () ) {
  case MediaType::AUDIO:
    return KMS_ELEMENT_PAD_TYPE_AUDIO;

  case MediaType::VIDEO:
    return KMS_ELEMENT_PAD_TYPE_VIDEO;

  case MediaType::DATA:
    return KMS_ELEMENT_PAD_TYPE_DATA;
  }

  throw KurentoException (UNSUPPORTED_MEDIA_TYPE, "Usupported media type");
}

static bool
check_parent_element (GstObject *src, GstObject *elem)
{
  GstObject *parent = GST_OBJECT_PARENT ( src);

  if (src == elem) {
    return true;
  }

  while (parent != NULL) {
    if ( parent == elem ) {
      return true;
    }

    parent = GST_OBJECT_PARENT (parent);
  }

  return false;
}

void
_media_element_impl_bus_message (GstBus *bus, GstMessage *message,
                                 gpointer data)
{
  if (message->type == GST_MESSAGE_ERROR) {
    GError *err = NULL;
    gchar *debug = NULL;
    MediaElementImpl *elem = reinterpret_cast <MediaElementImpl *> (data);

    if (elem == NULL) {
      return;
    }

    if (!check_parent_element (message->src, GST_OBJECT (elem->element) ) ) {
      return;
    }

    GST_ERROR ("MediaElement error: %" GST_PTR_FORMAT, message);
    gst_message_parse_error (message, &err, &debug);
    std::string errorMessage;

    if (err) {
      errorMessage = std::string (err->message);
    }

    if (debug != NULL) {
      errorMessage += " -> " + std::string (debug);
    }

    try {
      gint code = 0;

      if (err) {
        code = err->code;
      }

      Error error (elem->shared_from_this(), errorMessage, code,
                   "UNEXPECTED_ELEMENT_ERROR");

      elem->signalError (error);
    } catch (std::bad_weak_ptr &e) {
    }

    g_error_free (err);
    g_free (debug);
  }
}

void
_media_element_pad_added (GstElement *elem, GstPad *pad, gpointer data)
{
  MediaElementImpl *self = (MediaElementImpl *) data;
  bool retry = false;

  GST_LOG_OBJECT (pad, "Pad added");

  do {
    if (retry) {
      GST_DEBUG_OBJECT (pad, "Retriying connection");
    }

    if (GST_PAD_IS_SRC (pad) ) {
      std::unique_lock<std::recursive_timed_mutex> lock (self->sinksMutex,
          std::defer_lock);
      std::shared_ptr<MediaType> type;

      retry = !lock.try_lock_for (std::chrono::milliseconds {self->dist (self->rnd) });

      if (retry) {
        continue;
      }

      std::string description;

      if (g_str_has_prefix (GST_OBJECT_NAME (pad), "audio_") ) {
        type = std::shared_ptr<MediaType> (new MediaType (MediaType::AUDIO) );
        description = std::string (GST_OBJECT_NAME (pad) + sizeof ("audio_src") );
      } else if (g_str_has_prefix (GST_OBJECT_NAME (pad), "video_") ) {
        type = std::shared_ptr<MediaType> (new MediaType (MediaType::VIDEO) );
        description = std::string (GST_OBJECT_NAME (pad) + sizeof ("video_src") );
      } else {
        type = std::shared_ptr<MediaType> (new MediaType (MediaType::DATA) );
        description = std::string (GST_OBJECT_NAME (pad) + sizeof ("data_src") );
      }

      auto pos = description.find_last_of ("_");
      description.erase (pos);

      try {
        auto connections = self->sinks.at (type).at (description);

        for (auto it : connections) {
          if (g_strcmp0 (GST_OBJECT_NAME (pad), it->getSourcePadName() ) == 0) {
            std::unique_lock<std::recursive_timed_mutex> sinkLock (
              it->getSink()->sourcesMutex,
              std::defer_lock);

            retry = !sinkLock.try_lock_for (std::chrono::milliseconds {self->dist (self->rnd) });

            if (retry) {
              continue;
            }

            self->performConnection (it);
          }
        }
      } catch (std::out_of_range) {

      }
    } else {
      std::unique_lock<std::recursive_timed_mutex> lock (self->sourcesMutex,
          std::defer_lock);
      std::shared_ptr<MediaType> type;

      retry = !lock.try_lock_for (std::chrono::milliseconds {self->dist (self->rnd) });

      if (retry) {
        continue;
      }

      std::string description;

      if (g_str_has_prefix (GST_OBJECT_NAME (pad), "sink_audio_") ) {
        type = std::shared_ptr<MediaType> (new MediaType (MediaType::AUDIO) );
        description = std::string (GST_OBJECT_NAME (pad) + sizeof ("sink_audio") );
      } else if (g_str_has_prefix (GST_OBJECT_NAME (pad), "sink_video_") ) {
        type = std::shared_ptr<MediaType> (new MediaType (MediaType::VIDEO) );
        description = std::string (GST_OBJECT_NAME (pad) + sizeof ("sink_video") );
      } else {
        type = std::shared_ptr<MediaType> (new MediaType (MediaType::DATA) );
        description = std::string (GST_OBJECT_NAME (pad) + sizeof ("sink_data") );
      }

      try {
        auto sourceData = self->sources.at (type).at (description);

        auto source = sourceData->getSource();

        if (source) {
          if (g_strcmp0 (GST_OBJECT_NAME (pad),
                         sourceData->getSinkPadName().c_str() ) == 0) {
            std::unique_lock<std::recursive_timed_mutex> sourceLock (source->sinksMutex,
                std::defer_lock);

            retry = !sourceLock.try_lock_for (std::chrono::milliseconds {self->dist (self->rnd) });

            if (retry) {
              continue;
            }

            source->performConnection (sourceData);
          }
        }
      } catch (std::out_of_range) {

      }
    }
  } while (retry);
}

std::string
padTypeToString (KmsElementPadType type)
{
  switch (type) {
  case KMS_ELEMENT_PAD_TYPE_AUDIO:
    return "audio";

  case KMS_ELEMENT_PAD_TYPE_VIDEO:
    return "video";

  default:
    return "undefined";
  }
}

std::shared_ptr<MediaType>
padTypeToMediaType (KmsElementPadType type)
{
  switch (type) {
  case KMS_ELEMENT_PAD_TYPE_AUDIO:
    return std::make_shared <MediaType> (MediaType::AUDIO);

  case KMS_ELEMENT_PAD_TYPE_VIDEO:
    return std::make_shared <MediaType> (MediaType::VIDEO);

  default:
    break;
  }

  throw KurentoException (UNSUPPORTED_MEDIA_TYPE, "Usupported media type");
}

void
MediaElementImpl::mediaFlowOutStateChange (gboolean isFlowing, gchar *padName,
    KmsElementPadType type)
{
  std::shared_ptr<MediaFlowState > state;
  std::shared_ptr<MediaFlowData > data;
  std::string key;
  std::map<std::string, std::shared_ptr <MediaFlowData>>::iterator it;

  if (isFlowing) {
    GST_DEBUG_OBJECT (element, "Media Flowing OUT in pad %s with type %s", padName,
                      padTypeToString (type).c_str () );
    state = std::make_shared <MediaFlowState> (MediaFlowState::FLOWING);
  } else {
    GST_DEBUG_OBJECT (element, "Media NOT Flowing OUT in pad %s with type %s",
                      padName, padTypeToString (type).c_str () );
    state = std::make_shared <MediaFlowState> (MediaFlowState::NOT_FLOWING);
  }

  if (type == KMS_ELEMENT_PAD_TYPE_VIDEO) {
    key = std::string (TYPE_VIDEO) + std::string (padName);
  } else {
    key = std::string (TYPE_AUDIO) + std::string (padName);
  }

  it = mediaFlowDataOut.find (key);

  if (it != mediaFlowDataOut.end() ) {
    it->second->setState (state);
  } else {
    data = std::make_shared <MediaFlowData> (padTypeToMediaType (type),
           std::string (padName), state);
    mediaFlowDataOut[key] = data;
  }

  try {
    MediaFlowOutStateChange event (shared_from_this(),
                                   MediaFlowOutStateChange::getName (),
                                   state, padName, padTypeToMediaType (type) );

    signalMediaFlowOutStateChange (event);
  } catch (std::bad_weak_ptr &e) {
  }
}

void
MediaElementImpl::mediaFlowInStateChange (gboolean isFlowing, gchar *padName,
    KmsElementPadType type)
{
  std::shared_ptr<MediaFlowState > state;
  std::shared_ptr<MediaFlowData > data;
  std::string key;
  std::map<std::string, std::shared_ptr <MediaFlowData>>::iterator it;

  if (isFlowing) {
    GST_DEBUG_OBJECT (element, "Media Flowing IN in pad %s with type %s", padName,
                      padTypeToString (type).c_str () );
    state = std::make_shared <MediaFlowState> (MediaFlowState::FLOWING);
  } else {
    GST_DEBUG_OBJECT (element, "Media NOT Flowing IN in pad %s with type %s",
                      padName, padTypeToString (type).c_str () );
    state = std::make_shared <MediaFlowState> (MediaFlowState::NOT_FLOWING);
  }

  if (type == KMS_ELEMENT_PAD_TYPE_VIDEO) {
    key = std::string (TYPE_VIDEO) + std::string (padName);
  } else {
    key = std::string (TYPE_AUDIO) + std::string (padName);
  }

  it = mediaFlowDataIn.find (key);

  if (it != mediaFlowDataIn.end() ) {
    it->second->setState (state);
  } else {
    data = std::make_shared <MediaFlowData> (padTypeToMediaType (type),
           std::string (padName), state);
    mediaFlowDataIn[key] = data;
  }

  try {
    MediaFlowInStateChange event (shared_from_this(),
                                  MediaFlowInStateChange::getName (),
                                  state, padName, padTypeToMediaType (type) );

    signalMediaFlowInStateChange (event);
  } catch (std::bad_weak_ptr &e) {
  }
}

void
MediaElementImpl::postConstructor ()
{
  MediaObjectImpl::postConstructor ();

  mediaFlowOutHandler = register_signal_handler (G_OBJECT (element),
                        "flow-out-media",
                        std::function <void (GstElement *, gboolean, gchar *, KmsElementPadType) >
                        (std::bind (&MediaElementImpl::mediaFlowOutStateChange, this,
                                    std::placeholders::_2, std::placeholders::_3, std::placeholders::_4) ),
                        std::dynamic_pointer_cast<MediaElementImpl>
                        (shared_from_this() ) );

  mediaFlowInHandler = register_signal_handler (G_OBJECT (element),
                       "flow-in-media",
                       std::function <void (GstElement *, gboolean, gchar *, KmsElementPadType) >
                       (std::bind (&MediaElementImpl::mediaFlowInStateChange, this,
                                   std::placeholders::_2, std::placeholders::_3, std::placeholders::_4) ),
                       std::dynamic_pointer_cast<MediaElementImpl>
                       (shared_from_this() ) );
}

MediaElementImpl::MediaElementImpl (const boost::property_tree::ptree &config,
                                    std::shared_ptr<MediaObjectImpl> parent,
                                    const std::string &factoryName) : MediaObjectImpl (config, parent)
{
  std::shared_ptr<MediaPipelineImpl> pipe;

  pipe = std::dynamic_pointer_cast<MediaPipelineImpl> (getMediaPipeline() );

  element = gst_element_factory_make (factoryName.c_str(), NULL);

  if (element == NULL) {
    throw KurentoException (MEDIA_OBJECT_NOT_AVAILABLE,
                            "Cannot create gstreamer element: " + factoryName);
  }

  bus = gst_pipeline_get_bus (GST_PIPELINE (pipe->getPipeline () ) );
  handlerId = g_signal_connect (bus, "message",
                                G_CALLBACK (_media_element_impl_bus_message), this);

  padAddedHandlerId = g_signal_connect (element, "pad_added",
                                        G_CALLBACK (_media_element_pad_added), this);

  g_object_ref (element);
  pipe->addElement (element);

  //read default configuration for output bitrate
  try {
    int bitrate = getConfigValue<int, MediaElement> ("outputBitrate");
    GST_DEBUG ("Output bitrate configured to %d bps", bitrate);
    g_object_set (G_OBJECT (element), MIN_OUTPUT_BITRATE, bitrate,
                  MAX_OUTPUT_BITRATE, bitrate, NULL);
  } catch (boost::property_tree::ptree_error &e) {
  }

}

MediaElementImpl::~MediaElementImpl ()
{
  std::shared_ptr<MediaPipelineImpl> pipe;

  GST_LOG ("Deleting media element %s", getName().c_str () );

  if (padAddedHandlerId) {
    g_signal_handler_disconnect (element, padAddedHandlerId);
  }

  if (mediaFlowOutHandler > 0) {
    unregister_signal_handler (element, mediaFlowOutHandler);
  }

  if (mediaFlowInHandler > 0) {
    unregister_signal_handler (element, mediaFlowInHandler);
  }

  disconnectAll();

  pipe = std::dynamic_pointer_cast<MediaPipelineImpl> (getMediaPipeline() );

  gst_element_set_locked_state (element, TRUE);
  gst_element_set_state (element, GST_STATE_NULL);
  gst_bin_remove (GST_BIN ( pipe->getPipeline() ), element);

  g_object_unref (element);

  g_signal_handler_disconnect (bus, handlerId);
  g_object_unref (bus);
}

void
MediaElementImpl::release ()
{
  disconnectAll ();

  MediaObjectImpl::release();
}

void MediaElementImpl::disconnectAll ()
{
  while (!getSinkConnections().empty() ) {
    std::unique_lock<std::recursive_timed_mutex> sinkLock (sinksMutex,
        std::defer_lock);

    if (!sinkLock.try_lock_for (std::chrono::milliseconds {dist (rnd) }) ) {
      GST_DEBUG_OBJECT (getGstreamerElement(), "Retry disconnect all");
      continue;
    }

    for (std::shared_ptr<ElementConnectionData> connData : getSinkConnections() ) {
      auto sinkImpl = std::dynamic_pointer_cast <MediaElementImpl>
                      (connData->getSink () );
      std::unique_lock<std::recursive_timed_mutex> sinkLock (sinkImpl->sourcesMutex,
          std::defer_lock);

      if (sinkLock.try_lock_for (std::chrono::milliseconds {dist (rnd) }) ) {
        disconnect (connData->getSink (), connData->getType (),
                    connData->getSourceDescription (),
                    connData->getSinkDescription () );
      }
      else {
        GST_DEBUG_OBJECT (sinkImpl->getGstreamerElement(),
                          "Retry disconnect all %" GST_PTR_FORMAT, getGstreamerElement() );
      }
    }
  }

  while (!getSourceConnections().empty() ) {
    std::unique_lock<std::recursive_timed_mutex> sourceLock (sourcesMutex,
        std::defer_lock);

    if (!sourceLock.try_lock_for (std::chrono::milliseconds {dist (rnd) }) ) {
      GST_DEBUG_OBJECT (getGstreamerElement(), "Retry disconnect all");
      continue;
    }

    for (std::shared_ptr<ElementConnectionData> connData :
         getSourceConnections() ) {
      auto sourceImpl = std::dynamic_pointer_cast <MediaElementImpl>
                        (connData->getSource () );
      std::unique_lock<std::recursive_timed_mutex> sourceLock (sourceImpl->sinksMutex,
          std::defer_lock);

      if (sourceLock.try_lock_for (std::chrono::milliseconds {dist (rnd) }) ) {
        connData->getSource ()->disconnect (connData->getSink (),
                                            connData->getType (),
                                            connData->getSourceDescription (),
                                            connData->getSinkDescription () );
      }
      else {
        GST_DEBUG_OBJECT (sourceImpl->getGstreamerElement (),
                          "Retry disconnect all %" GST_PTR_FORMAT, getGstreamerElement() );
      }
    }
  }
}

std::vector<std::shared_ptr<ElementConnectionData>>
    MediaElementImpl::getSourceConnections ()
{
  std::unique_lock<std::recursive_timed_mutex> lock (sourcesMutex);
  std::vector<std::shared_ptr<ElementConnectionData>> ret;

  for (auto it : sources) {
    for (auto it2 : it.second) {
      try {
        ret.push_back (it2.second->toInterface() );
      } catch (KurentoException) {
      }
    }
  }

  return ret;
}

std::vector<std::shared_ptr<ElementConnectionData>>
    MediaElementImpl::getSourceConnections (
      std::shared_ptr<MediaType> mediaType)
{
  std::unique_lock<std::recursive_timed_mutex> lock (sourcesMutex);
  std::vector<std::shared_ptr<ElementConnectionData>> ret;

  try {
    for (auto it : sources.at (mediaType) ) {
      try {
        ret.push_back (it.second->toInterface() );
      } catch (KurentoException) {

      }
    }
  } catch (std::out_of_range) {

  }

  return ret;
}

std::vector<std::shared_ptr<ElementConnectionData>>
    MediaElementImpl::getSourceConnections (
      std::shared_ptr<MediaType> mediaType, const std::string &description)
{
  std::unique_lock<std::recursive_timed_mutex> lock (sourcesMutex);
  std::vector<std::shared_ptr<ElementConnectionData>> ret;

  try {
    ret.push_back (sources.at (mediaType).at (description)->toInterface() );
  } catch (KurentoException) {

  } catch (std::out_of_range) {

  }

  return ret;
}

std::vector<std::shared_ptr<ElementConnectionData>>
    MediaElementImpl::getSinkConnections ()
{
  std::unique_lock<std::recursive_timed_mutex> lock (sinksMutex);
  std::vector<std::shared_ptr<ElementConnectionData>> ret;

  for (auto it : sinks) {
    for (auto it2 : it.second) {
      for (auto it3 : it2.second) {
        try {
          ret.push_back (it3->toInterface() );
        } catch (KurentoException) {

        }
      }
    }
  }

  return ret;
}

std::vector<std::shared_ptr<ElementConnectionData>>
    MediaElementImpl::getSinkConnections (
      std::shared_ptr<MediaType> mediaType)
{
  std::unique_lock<std::recursive_timed_mutex> lock (sinksMutex);
  std::vector<std::shared_ptr<ElementConnectionData>> ret;

  try {
    for (auto it : sinks.at (mediaType) ) {
      for (auto it3 : it.second) {
        try {
          ret.push_back (it3->toInterface() );
        } catch (KurentoException) {

        }
      }
    }
  } catch (std::out_of_range) {

  }

  return ret;
}

std::vector<std::shared_ptr<ElementConnectionData>>
    MediaElementImpl::getSinkConnections (
      std::shared_ptr<MediaType> mediaType, const std::string &description)
{
  std::unique_lock<std::recursive_timed_mutex> lock (sinksMutex);
  std::vector<std::shared_ptr<ElementConnectionData>> ret;

  try {
    for (auto it : sinks.at (mediaType).at (description) ) {
      try {
        ret.push_back (it->toInterface() );
      } catch (KurentoException) {

      }
    }
  } catch (std::out_of_range) {

  }

  return ret;
}

void MediaElementImpl::connect (std::shared_ptr<MediaElement> sink)
{
  // Until mediaDescriptions are really used, we just connect audio an video
  connect (sink, std::shared_ptr<MediaType> (new MediaType (MediaType::AUDIO) ),
           DEFAULT, DEFAULT);
  connect (sink, std::shared_ptr<MediaType> (new MediaType (MediaType::VIDEO) ),
           DEFAULT, DEFAULT);
  connect (sink, std::shared_ptr<MediaType> (new MediaType (MediaType::DATA) ),
           DEFAULT, DEFAULT);
}

void MediaElementImpl::connect (std::shared_ptr<MediaElement> sink,
                                std::shared_ptr<MediaType> mediaType)
{
  connect (sink, mediaType, DEFAULT, DEFAULT);
}

void MediaElementImpl::connect (std::shared_ptr<MediaElement> sink,
                                std::shared_ptr<MediaType> mediaType,
                                const std::string &sourceMediaDescription)
{
  connect (sink, mediaType, sourceMediaDescription, DEFAULT);
}

void MediaElementImpl::prepareSinkConnection (std::shared_ptr<MediaElement> src,
    std::shared_ptr< MediaType > mediaType,
    const std::string &sourceMediaDescription,
    const std::string &sinkMediaDescription)
{
  /* Do nothing by default */
}

void MediaElementImpl::connect (std::shared_ptr<MediaElement> sink,
                                std::shared_ptr<MediaType> mediaType,
                                const std::string &sourceMediaDescription,
                                const std::string &sinkMediaDescription)
{
  KmsElementPadType type;
  gchar *padName;
  std::shared_ptr<MediaElementImpl> sinkImpl =
    std::dynamic_pointer_cast<MediaElementImpl> (sink);

  if (sinkImpl->getMediaPipeline ()->getId () != getMediaPipeline ()->getId() ) {
    throw KurentoException (CONNECT_ERROR,
                            "Media elements do not share pipeline");
  }

  std::unique_lock<std::recursive_timed_mutex> lock (sinksMutex);
  std::unique_lock<std::recursive_timed_mutex> sinkLock (sinkImpl->sourcesMutex);
  std::vector <std::shared_ptr <ElementConnectionData>> connections;
  std::shared_ptr <ElementConnectionDataInternal> connectionData (
    new ElementConnectionDataInternal (std::dynamic_pointer_cast<MediaElement>
                                       (shared_from_this () ), sink, mediaType,
                                       sourceMediaDescription,
                                       sinkMediaDescription) );

  GST_DEBUG ("Connecting %s -> %s params %s %s %s", getName().c_str(),
             sink->getName ().c_str (), mediaType->getString ().c_str (),
             sourceMediaDescription.c_str(), sinkMediaDescription.c_str() );

  connections = sink->getSourceConnections (mediaType, sinkMediaDescription);

  if (!connections.empty () ) {
    std::shared_ptr <ElementConnectionData> connection = connections.at (0);
    connection->getSource()->disconnect (connection->getSink (), mediaType,
                                         sourceMediaDescription,
                                         connection->getSinkDescription () );
  }

  sinkImpl->prepareSinkConnection (connectionData->getSource(), mediaType,
                                   sourceMediaDescription, sinkMediaDescription);

  type = convertMediaType (mediaType);
  g_signal_emit_by_name (getGstreamerElement (), "request-new-pad", type,
                         sourceMediaDescription.c_str (), GST_PAD_SRC, &padName,
                         NULL);

  if (padName == NULL) {
    throw KurentoException (CONNECT_ERROR, "Element: '" + getName() +
                            "'does note provide a connection for " +
                            mediaType->getString () + "-" +
                            sourceMediaDescription);
  }

  connectionData->setSourcePadName (padName);

  sinks[mediaType][sourceMediaDescription].insert (connectionData);
  sinkImpl->sources[mediaType][sinkMediaDescription] = connectionData;

  performConnection (connectionData);

  sinkLock.unlock();
  lock.unlock ();

  ElementConnected elementConnected (shared_from_this(),
                                     ElementConnected::getName (),
                                     sink, mediaType, sourceMediaDescription,
                                     sinkMediaDescription);
  signalElementConnected (elementConnected);
}

void
MediaElementImpl::performConnection (std::shared_ptr
                                     <ElementConnectionDataInternal> data)
{
  GstPad *src = NULL, *sink = NULL;

  src = data->getSourcePad ();

  if (!src) {
    GST_LOG ("Still waiting for src pad %s:%s", getName().c_str(),
             data->getSourcePadName () );
    return;
  }

  sink = data->getSinkPad ();

  if (sink) {
    GstPadLinkReturn ret;

    GST_LOG ("Linking %s:%s -> %s:%s", getName().c_str(),
             data->getSourcePadName (), data->getSink()->getName().c_str(),
             data->getSinkPadName ().c_str() );

    ret = gst_pad_link_full (src, sink, GST_PAD_LINK_CHECK_NOTHING);

    if (ret != GST_PAD_LINK_OK) {
      GST_WARNING ("Cannot link pads: %" GST_PTR_FORMAT " and %" GST_PTR_FORMAT
                   " reason: %s", src, sink, gst_pad_link_get_name (ret) );
    } else {
      GST_LOG ("Link done");
    }

    g_object_unref (sink);
  } else {
    GST_LOG ("Still waiting for sink pad %s:%s",
             data->getSink()->getName().c_str(),
             data->getSinkPadName ().c_str() );
  }

  g_object_unref (src);
}

void MediaElementImpl::disconnect (std::shared_ptr<MediaElement> sink)
{
  // Until mediaDescriptions are really used, we just disconnect audio an video
  disconnect (sink, std::shared_ptr<MediaType> (new MediaType (
                MediaType::AUDIO) ), DEFAULT, DEFAULT);
  disconnect (sink, std::shared_ptr<MediaType> (new MediaType (
                MediaType::VIDEO) ), DEFAULT, DEFAULT);
  disconnect (sink, std::shared_ptr<MediaType> (new MediaType (
                MediaType::DATA) ), DEFAULT, DEFAULT);
}

void MediaElementImpl::disconnect (std::shared_ptr<MediaElement> sink,
                                   std::shared_ptr<MediaType> mediaType)
{
  disconnect (sink, mediaType, DEFAULT, DEFAULT);
}

void MediaElementImpl::disconnect (std::shared_ptr<MediaElement> sink,
                                   std::shared_ptr<MediaType> mediaType,
                                   const std::string &sourceMediaDescription)
{
  disconnect (sink, mediaType, sourceMediaDescription, DEFAULT);
}

void MediaElementImpl::disconnect (std::shared_ptr<MediaElement> sink,
                                   std::shared_ptr<MediaType> mediaType,
                                   const std::string &sourceMediaDescription,
                                   const std::string &sinkMediaDescription)
{
  if (!sink) {
    GST_WARNING ("Sink not available while disconnecting");
    return;
  }

  std::shared_ptr<MediaElementImpl> sinkImpl =
    std::dynamic_pointer_cast<MediaElementImpl> (sink);
  std::unique_lock<std::recursive_timed_mutex> sinkLock (sinkImpl->sourcesMutex);
  std::unique_lock<std::recursive_timed_mutex> lock (sinksMutex);

  GST_DEBUG ("Disconnecting %s - %s params %s %s %s", getName().c_str(),
             sink->getName ().c_str (), mediaType->getString ().c_str (),
             sourceMediaDescription.c_str(), sinkMediaDescription.c_str() );

  try {
    std::shared_ptr<ElementConnectionDataInternal> connectionData;
    gboolean ret;

    connectionData = sinkImpl->sources.at (mediaType).at (sinkMediaDescription);

    if (connectionData->toInterface()->getSourceDescription() ==
        sourceMediaDescription) {
      sinkImpl->sources.at (mediaType).erase (sinkMediaDescription);
    }

    for (auto conn : sinks.at (mediaType).at (sourceMediaDescription) ) {
      if (conn->toInterface()->getSink() == sink &&
          conn->toInterface()->getSinkDescription() == sinkMediaDescription) {
        sinks.at (mediaType).at (sourceMediaDescription).erase (conn);
        break;
      }
    }

    g_signal_emit_by_name (getGstreamerElement (), "release-requested-pad",
                           connectionData->getSourcePadName (), &ret, NULL);
  } catch (std::out_of_range) {

  }

  sinkLock.unlock();
  lock.unlock ();

  ElementDisconnected elementDisconnected (shared_from_this(),
      ElementDisconnected::getName (),
      sink, mediaType, sourceMediaDescription,
      sinkMediaDescription);
  signalElementDisconnected (elementDisconnected);
}

static void
append_codec_to_array (GArray *array, const char *codec)
{
  GValue v = G_VALUE_INIT;
  GstStructure *s;

  g_value_init (&v, GST_TYPE_STRUCTURE);
  s = gst_structure_new_empty (codec);
  gst_value_set_structure (&v, s);
  gst_structure_free (s);
  g_array_append_val (array, v);
}

void MediaElementImpl::setAudioFormat (std::shared_ptr<AudioCaps> caps)
{
  std::shared_ptr<AudioCodec> codec;
  std::stringstream sstm;
  std::string str_caps;
  GstCaps *c = NULL;
  GArray *audio_codecs;

  codec = caps->getCodec();

  switch (codec->getValue() ) {
  case AudioCodec::OPUS:
    str_caps = "audio/x-opus";
    audio_codecs_list.push_back ("opus/48000/2");
    break;

  case AudioCodec::PCMU:
    str_caps = "audio/x-mulaw";
    audio_codecs_list.push_back ("PCMU/8000");
    break;

  case AudioCodec::PCMA:
    str_caps = "audio/x-alaw";
    audio_codecs_list.push_back ("PCMA/8000");
    break;

  case AudioCodec::TELEPHONE_EVENT:
    str_caps = "audio/x-raw";
    audio_codecs_list.push_back ("telephone-event/8000");
    break;

  case AudioCodec::RAW:
    str_caps = "audio/x-raw";
    break;

  default:
    throw KurentoException (MEDIA_OBJECT_ILLEGAL_PARAM_ERROR,
                            "Invalid parameter provided: " + codec->getString() );
  }

  sstm << str_caps << ", bitrate=(int)" << caps->getBitrate();
  str_caps = sstm.str ();

  c = gst_caps_from_string (str_caps.c_str() );
  g_object_set (element, "audio-caps", c, NULL);

  audio_codecs = g_array_new (FALSE, TRUE, sizeof (GValue) ); //ru-bu

  //g_object_get(G_OBJECT(element), "audio-codecs", &audio_codecs, NULL);
  for (std::string audio_codecs_str : audio_codecs_list) {
    append_codec_to_array (audio_codecs, audio_codecs_str.c_str() );
  }

  if (audio_codecs_list.size() >= 1) {
    g_object_set (element, "num-audio-medias", 1, NULL);
    g_object_set (G_OBJECT (element), "audio-codecs", audio_codecs, NULL);
  } else {
    g_array_free (audio_codecs, TRUE);
  }

}

void MediaElementImpl::setVideoFormat (std::shared_ptr<VideoCaps> caps)
{
  std::shared_ptr<VideoCodec> codec;
  std::shared_ptr<Fraction> fraction;
  std::stringstream sstm;
  std::string str_caps;
  GstCaps *c = NULL;
  GArray *video_codecs;

  codec = caps->getCodec();
  fraction = caps->getFramerate();

  switch (codec->getValue() ) {
  case VideoCodec::VP8:
    str_caps = "video/x-vp8";
    video_codecs_list.push_back ("VP8/90000");
    break;

  case VideoCodec::H264:
    str_caps = "video/x-h264";
    video_codecs_list.push_back ("H264/90000");
    break;

  case VideoCodec::RAW:
    str_caps = "video/x-raw";
    //video_codecs_list.push_back("");
    break;

  default:
    throw KurentoException (MEDIA_OBJECT_ILLEGAL_PARAM_ERROR,
                            "Invalid parameter provided: " + codec->getString() );
  }

  sstm << str_caps << ", framerate=(fraction)" << fraction->getNumerator() <<
       "/" << fraction->getDenominator();
  str_caps = sstm.str ();

  c = gst_caps_from_string (str_caps.c_str() );
  g_object_set (element, "video-caps", c, NULL);

  video_codecs = g_array_new (FALSE, TRUE, sizeof (GValue) );

  for (std::string video_codecs_str : video_codecs_list) {
    append_codec_to_array (video_codecs, video_codecs_str.c_str() );
  }

  if (video_codecs_list.size() >= 1) {
    g_object_set (element, "num-video-medias", 1, NULL);
    g_object_set (G_OBJECT (element), "video-codecs", video_codecs, NULL);
  } else {
    g_array_free (video_codecs, TRUE);
  }
}

std::string MediaElementImpl::getGstreamerDot (
  std::shared_ptr<GstreamerDotDetails> details)
{
  return generateDotGraph (GST_BIN (element), details);
}

std::string MediaElementImpl::getGstreamerDot()
{
  return generateDotGraph (GST_BIN (element),
                           std::shared_ptr <GstreamerDotDetails> (new GstreamerDotDetails (
                                 GstreamerDotDetails::SHOW_VERBOSE) ) );
}

void MediaElementImpl::setOutputBitrate (int bitrate)
{
  GST_WARNING ("setOutputBitrate method is deprecated, use properties "
               "minOutputBitrate and maxOutputBitrate");
  g_object_set (G_OBJECT (element), MIN_OUTPUT_BITRATE, bitrate,
                MAX_OUTPUT_BITRATE, bitrate, NULL);
}

int MediaElementImpl::getMinOuputBitrate ()
{
  gint bitrate;

  GST_WARNING ("minOuputBitrate property is deprecated, use property "
               "minOutputBitrate");
  g_object_get (G_OBJECT (element), MIN_OUTPUT_BITRATE, &bitrate, NULL);

  return bitrate;
}

int MediaElementImpl::getMinOutputBitrate ()
{
  gint bitrate;

  g_object_get (G_OBJECT (element), MIN_OUTPUT_BITRATE, &bitrate, NULL);

  return bitrate;
}

void MediaElementImpl::setMinOuputBitrate (int minOuputBitrate)
{
  GST_WARNING ("minOuputBitrate property is deprecated, use property "
               "minOutputBitrate");
  g_object_set (G_OBJECT (element), MIN_OUTPUT_BITRATE, minOuputBitrate,
                NULL);
}

void MediaElementImpl::setMinOutputBitrate (int minOutputBitrate)
{
  g_object_set (G_OBJECT (element), MIN_OUTPUT_BITRATE, minOutputBitrate,
                NULL);
}

int MediaElementImpl::getMaxOuputBitrate ()
{
  gint bitrate;

  GST_WARNING ("maxOuputBitrate property is deprecated, use property "
               "maxOutputBitrate");
  g_object_get (G_OBJECT (element), MAX_OUTPUT_BITRATE, &bitrate, NULL);

  return bitrate;
}

int MediaElementImpl::getMaxOutputBitrate ()
{
  gint bitrate;

  g_object_get (G_OBJECT (element), MAX_OUTPUT_BITRATE, &bitrate, NULL);

  return bitrate;
}

void MediaElementImpl::setMaxOuputBitrate (int maxOuputBitrate)
{
  GST_WARNING ("maxOuputBitrate property is deprecated, use property "
               "maxOutputBitrate");
  g_object_set (G_OBJECT (element), MAX_OUTPUT_BITRATE, maxOuputBitrate,
                NULL);
}

void MediaElementImpl::setMaxOutputBitrate (int maxOutputBitrate)
{
  g_object_set (G_OBJECT (element), MAX_OUTPUT_BITRATE, maxOutputBitrate,
                NULL);
}

std::map <std::string, std::shared_ptr<Stats>>
    MediaElementImpl::generateStats (const gchar *selector)
{
  std::map <std::string, std::shared_ptr<Stats>> statsReport;
  GstStructure *stats;

  g_signal_emit_by_name (getGstreamerElement(), "stats", selector, &stats);

  fillStatsReport (statsReport, stats, time (NULL) );

  gst_structure_free (stats);

  return statsReport;
}

std::map <std::string, std::shared_ptr<Stats>>
    MediaElementImpl::getStats ()
{
  return generateStats (NULL);
}

std::map <std::string, std::shared_ptr<Stats>>
    MediaElementImpl::getStats (std::shared_ptr<MediaType> mediaType)
{
  const gchar *selector = NULL;

  switch (mediaType->getValue () ) {
  case MediaType::AUDIO:
    selector = "audio";
    break;

  case MediaType::VIDEO:
    selector = "video";
    break;

  case MediaType::DATA:
    selector = "data";
    break;

  default:
    throw KurentoException (MEDIA_OBJECT_ILLEGAL_PARAM_ERROR,
                            "Unsupported media type: " + mediaType->getString() );
  }

  return generateStats (selector);
}

static std::shared_ptr<MediaType>
getMediaTypeFromTypeSelector (const gchar *type)
{
  if (g_strcmp0 (type, "audio") == 0) {
    return std::make_shared <MediaType> (MediaType::AUDIO);
  } else if (g_strcmp0 (type, "video") == 0) {
    return std::make_shared <MediaType> (MediaType::VIDEO);
  } else {
    return std::make_shared <MediaType> (MediaType::DATA);
  }
}

void
MediaElementImpl::collectLatencyStats (
  std::vector<std::shared_ptr<MediaLatencyStat>> &latencyStats,
  const GstStructure *stats)
{
  gint i, fields;

  fields = gst_structure_n_fields (stats);

  for (i = 0; i < fields; i ++) {
    const gchar *fieldname;
    const GValue *val;
    gchar *mediaType;
    guint64 avg;

    fieldname = gst_structure_nth_field_name (stats, i);
    val = gst_structure_get_value (stats, fieldname);

    if (!GST_VALUE_HOLDS_STRUCTURE (val) ) {
      GST_DEBUG ("Ignore unexpected value for field %s", fieldname);
      continue;
    }

    gst_structure_get (gst_value_get_structure (val), "type", G_TYPE_STRING,
                       &mediaType, "avg", G_TYPE_UINT64, &avg, NULL);

    std::shared_ptr<MediaType> type = getMediaTypeFromTypeSelector (mediaType);
    std::shared_ptr<MediaLatencyStat> latency =
      std::make_shared <MediaLatencyStat> (fieldname, type, avg);
    g_free (mediaType);

    latencyStats.push_back (latency);
  }
}

static void
setDeprecatedProperties (std::shared_ptr<ElementStats> eStats)
{
  std::vector<std::shared_ptr<MediaLatencyStat>> inStats =
        eStats->getInputLatency();

  for (unsigned i = 0; i < inStats.size(); i++) {
    if (inStats[i]->getName() == "sink_audio_default") {
      eStats->setInputAudioLatency (inStats[i]->getAvg() );
    } else if (inStats[i]->getName() == "sink_video_default") {
      eStats->setInputVideoLatency (inStats[i]->getAvg() );
    }
  }
}

void
MediaElementImpl::fillStatsReport (std::map
                                   <std::string, std::shared_ptr<Stats>>
                                   &report, const GstStructure *stats, double timestamp)
{
  std::shared_ptr<Stats> elementStats;
  GstStructure *latencies;
  const GValue *value;

  value = gst_structure_get_value (stats, KMS_MEDIA_ELEMENT_FIELD);

  if (value == NULL) {
    /* No element stats available */
    return;
  }

  if (!GST_VALUE_HOLDS_STRUCTURE (value) ) {
    gchar *str_val;

    str_val = g_strdup_value_contents (value);
    GST_WARNING ("Unexpected field type (%s) = %s", KMS_MEDIA_ELEMENT_FIELD,
                 str_val);
    g_free (str_val);

    return;
  }

  std::vector<std::shared_ptr<MediaLatencyStat>> inputLatencies;

  if (gst_structure_get (gst_value_get_structure (value), "input-latencies",
                         GST_TYPE_STRUCTURE, &latencies, NULL) ) {
    collectLatencyStats (inputLatencies, latencies);
    gst_structure_free (latencies);
  }

  if (report.find (getId () ) != report.end() ) {
    std::shared_ptr<ElementStats> eStats =
      std::dynamic_pointer_cast <ElementStats> (report[getId ()]);
    eStats->setInputLatency (inputLatencies);
  } else {
    elementStats = std::make_shared <ElementStats> (getId (),
                   std::make_shared <StatsType> (StatsType::element), timestamp,
                   0.0, 0.0, inputLatencies);
    report[getId ()] = elementStats;
  }

  setDeprecatedProperties (std::dynamic_pointer_cast <ElementStats>
                           (report[getId ()]) );
}

bool MediaElementImpl::isMediaFlowingIn (std::shared_ptr<MediaType> mediaType)
{
  return isMediaFlowingIn (mediaType, KMS_DEFAULT_MEDIA_DESCRIPTION);
}

bool MediaElementImpl::isMediaFlowingIn (std::shared_ptr<MediaType> mediaType,
    const std::string &sinkMediaDescription)
{
  std::string key;
  std::map<std::string, std::shared_ptr <MediaFlowData>>::iterator it;
  gboolean ret = false;

  if (mediaType->getValue () == MediaType::VIDEO) {
    key = std::string (TYPE_VIDEO) + std::string (sinkMediaDescription);
  } else if (mediaType->getValue () == MediaType::AUDIO) {
    key = std::string (TYPE_AUDIO) + std::string (sinkMediaDescription);
  } else {
    GST_ERROR ("Media type DATA is not supported for MediaFlowingIn");
    throw KurentoException (MEDIA_OBJECT_ILLEGAL_PARAM_ERROR,
                            "Media type DATA is not supported for MediaFlowingIn");
  }

  it = mediaFlowDataIn.find (key);

  if (it != mediaFlowDataIn.end() ) {
    if (it->second->getState ()->getValue () == MediaFlowState::FLOWING) {
      ret = true;
    }
  }

  return ret;
}

bool MediaElementImpl::isMediaFlowingOut (std::shared_ptr<MediaType> mediaType)
{
  return isMediaFlowingOut (mediaType, KMS_DEFAULT_MEDIA_DESCRIPTION);
}

bool MediaElementImpl::isMediaFlowingOut (std::shared_ptr<MediaType> mediaType,
    const std::string &sourceMediaDescription)
{
  std::string key;
  std::map<std::string, std::shared_ptr <MediaFlowData>>::iterator it;
  gboolean ret = false;

  if (mediaType->getValue () == MediaType::VIDEO) {
    key = std::string (TYPE_VIDEO) + std::string (sourceMediaDescription);
  } else if (mediaType->getValue () == MediaType::AUDIO) {
    key = std::string (TYPE_AUDIO) + std::string (sourceMediaDescription);
  } else {
    GST_ERROR ("Media type DATA is not supported for MediaFlowingOut");
    throw KurentoException (MEDIA_OBJECT_ILLEGAL_PARAM_ERROR,
                            "Media type DATA is not supported for MediaFlowingOut");
  }

  it = mediaFlowDataOut.find (key);

  if (it != mediaFlowDataOut.end() ) {
    if (it->second->getState ()->getValue () == MediaFlowState::FLOWING) {
      ret = true;
    }
  }

  return ret;
}

MediaElementImpl::StaticConstructor MediaElementImpl::staticConstructor;

MediaElementImpl::StaticConstructor::StaticConstructor()
{
  GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, GST_DEFAULT_NAME, 0,
                           GST_DEFAULT_NAME);
}

} /* kurento */


text:
import gql from 'graphql-tag';
import fetchMock from 'fetch-mock';
import { ASTNode, print, stripIgnoredCharacters } from 'graphql';
import { TextDecoder } from 'util';
import { ReadableStream } from 'web-streams-polyfill/ponyfill/es2018';
import { Readable } from 'stream';

import { Observable, Observer, ObservableSubscription } from '../../../utilities/observables/Observable';
import { ApolloLink } from '../../core/ApolloLink';
import { execute } from '../../core/execute';
import { HttpLink } from '../HttpLink';
import { createHttpLink } from '../createHttpLink';
import { ClientParseError } from '../serializeFetchParameter';
import { ServerParseError } from '../parseAndCheckHttpResponse';
import { ServerError } from '../../..';
import { voidFetchDuringEachTest } from './helpers';
import { itAsync } from '../../../testing';

const sampleQuery = gql`
  query SampleQuery {
    stub {
      id
    }
  }
`;

const sampleMutation = gql`
  mutation SampleMutation {
    stub {
      id
    }
  }
`;

const sampleDeferredQuery = gql`
  query SampleDeferredQuery {
    stub {
      id
      ... on Stub @defer {
        name
      }
    }
  }
`;

const sampleQueryCustomDirective = gql`
  query SampleDeferredQuery {
    stub {
      id
      ... on Stub @deferCustomDirective {
        name
      }
    }
  }
`;

function makeCallback<TArgs extends any[]>(
  resolve: () => void,
  reject: (error: Error) => void,
  callback: (...args: TArgs) => any,
) {
  return function () {
    try {
      callback.apply(this, arguments);
      resolve();
    } catch (error) {
      reject(error);
    }
  } as typeof callback;
}

function convertBatchedBody(body: BodyInit | null | undefined) {
  return JSON.parse(body as string);
}

function makePromise(res: any) {
  return new Promise((resolve) => setTimeout(() => resolve(res)));
}

describe('HttpLink', () => {
  describe('General', () => {
    const data = { data: { hello: 'world' } };
    const data2 = { data: { hello: 'everyone' } };
    const mockError = { throws: new TypeError('mock me') };
    let subscriber: Observer<any>;
    const subscriptions = new Set<ObservableSubscription>();

    beforeEach(() => {
      fetchMock.restore();
      fetchMock.post('begin:/data2', makePromise(data2));
      fetchMock.post('begin:/data', makePromise(data));
      fetchMock.post('begin:/error', mockError);
      fetchMock.post('begin:/apollo', makePromise(data));

      fetchMock.get('begin:/data', makePromise(data));
      fetchMock.get('begin:/data2', makePromise(data2));

      const next = jest.fn();
      const error = jest.fn();
      const complete = jest.fn();

      subscriber = {
        next,
        error,
        complete,
      };

      subscriptions.clear();
    });

    afterEach(() => {
      fetchMock.restore();
      if (subscriptions.size) {
        // Tests within this describe block can add subscriptions to this Set
        // that they want to be canceled/unsubscribed after the test finishes.
        subscriptions.forEach(sub => sub.unsubscribe());
      }
    });

    it('does not need any constructor arguments', () => {
      expect(() => new HttpLink()).not.toThrow();
    });

    itAsync('constructor creates link that can call next and then complete', (resolve, reject) => {
      const next = jest.fn();
      const link = new HttpLink({ uri: '/data' });
      const observable = execute(link, {
        query: sampleQuery,
      });
      observable.subscribe({
        next,
        error: error => expect(false),
        complete: () => {
          expect(next).toHaveBeenCalledTimes(1);
          resolve();
        },
      });
    });

    itAsync('supports using a GET request', (resolve, reject) => {
      const variables = { params: 'stub' };
      const extensions = { myExtension: 'foo' };

      const link = createHttpLink({
        uri: '/data',
        fetchOptions: { method: 'GET' },
        includeExtensions: true,
        includeUnusedVariables: true,
      });

      execute(link, { query: sampleQuery, variables, extensions }).subscribe({
        next: makeCallback(resolve, reject, () => {
          const [uri, options] = fetchMock.lastCall()!;
          const { method, body } = options!;
          expect(body).toBeUndefined();
          expect(method).toBe('GET');
          expect(uri).toBe(
            '/data?query=query%20SampleQuery%20%7B%0A%20%20stub%20%7B%0A%20%20%20%20id%0A%20%20%7D%0A%7D&operationName=SampleQuery&variables=%7B%22params%22%3A%22stub%22%7D&extensions=%7B%22myExtension%22%3A%22foo%22%7D',
          );
        }),
        error: error => reject(error),
      });
    });

    itAsync('supports using a GET request with search', (resolve, reject) => {
      const variables = { params: 'stub' };

      const link = createHttpLink({
        uri: '/data?foo=bar',
        fetchOptions: { method: 'GET' },
      });

      execute(link, { query: sampleQuery, variables }).subscribe({
        next: makeCallback(resolve, reject, () => {
          const [uri, options] = fetchMock.lastCall()!;
          const { method, body } = options!;
          expect(body).toBeUndefined();
          expect(method).toBe('GET');
          expect(uri).toBe(
            '/data?foo=bar&query=query%20SampleQuery%20%7B%0A%20%20stub%20%7B%0A%20%20%20%20id%0A%20%20%7D%0A%7D&operationName=SampleQuery&variables=%7B%7D',
          );
        }),
        error: error => reject(error),
      });
    });

    itAsync('supports using a GET request on the context', (resolve, reject) => {
      const variables = { params: 'stub' };
      const link = createHttpLink({
        uri: '/data',
      });

      execute(link, {
        query: sampleQuery,
        variables,
        context: {
          fetchOptions: { method: 'GET' },
        },
      }).subscribe(
        makeCallback(resolve, reject, () => {
          const [uri, options] = fetchMock.lastCall()!;
          const { method, body } = options!;
          expect(body).toBeUndefined();
          expect(method).toBe('GET');
          expect(uri).toBe(
            '/data?query=query%20SampleQuery%20%7B%0A%20%20stub%20%7B%0A%20%20%20%20id%0A%20%20%7D%0A%7D&operationName=SampleQuery&variables=%7B%7D',
          );
        }),
      );
    });

    itAsync('uses GET with useGETForQueries', (resolve, reject) => {
      const variables = { params: 'stub' };
      const link = createHttpLink({
        uri: '/data',
        useGETForQueries: true,
      });

      execute(link, {
        query: sampleQuery,
        variables,
      }).subscribe(
        makeCallback(resolve, reject, () => {
          const [uri, options] = fetchMock.lastCall()!;
          const { method, body } = options!;
          expect(body).toBeUndefined();
          expect(method).toBe('GET');
          expect(uri).toBe(
            '/data?query=query%20SampleQuery%20%7B%0A%20%20stub%20%7B%0A%20%20%20%20id%0A%20%20%7D%0A%7D&operationName=SampleQuery&variables=%7B%7D',
          );
        }),
      );
    });

    itAsync('uses POST for mutations with useGETForQueries', (resolve, reject) => {
      const variables = { params: 'stub' };
      const link = createHttpLink({
        uri: '/data',
        useGETForQueries: true,
      });

      execute(link, {
        query: sampleMutation,
        variables,
      }).subscribe(
        makeCallback(resolve, reject, () => {
          const [uri, options] = fetchMock.lastCall()!;
          const { method, body } = options!;
          expect(body).toBeDefined();
          expect(method).toBe('POST');
          expect(uri).toBe('/data');
        }),
      );
    });

    itAsync('strips unused variables, respecting nested fragments', (resolve, reject) => {
      const link = createHttpLink({ uri: '/data' });

      const query = gql`
        query PEOPLE (
          $declaredAndUsed: String,
          $declaredButUnused: Int,
        ) {
          people(
            surprise: $undeclared,
            noSurprise: $declaredAndUsed,
          ) {
            ... on Doctor {
              specialty(var: $usedByInlineFragment)
            }
            ...LawyerFragment
          }
        }
        fragment LawyerFragment on Lawyer {
          caseCount(var: $usedByNamedFragment)
        }
      `;

      const variables = {
        unused: 'strip',
        declaredButUnused: 'strip',
        declaredAndUsed: 'keep',
        undeclared: 'keep',
        usedByInlineFragment: 'keep',
        usedByNamedFragment: 'keep',
      };

      execute(link, {
        query,
        variables,
      }).subscribe({
        next: makeCallback(resolve, reject, () => {
          const [uri, options] = fetchMock.lastCall()!;
          const { method, body } = options!;
          expect(JSON.parse(body as string)).toEqual({
            operationName: "PEOPLE",
            query: print(query),
            variables: {
              declaredAndUsed: 'keep',
              undeclared: 'keep',
              usedByInlineFragment: 'keep',
              usedByNamedFragment: 'keep',
            },
          });
          expect(method).toBe('POST');
          expect(uri).toBe('/data');
        }),
        error: error => reject(error),
      });
    });

    itAsync('should add client awareness settings to request headers', (resolve, reject) => {
      const variables = { params: 'stub' };
      const link = createHttpLink({
        uri: '/data',
      });

      const clientAwareness = {
        name: 'Some Client Name',
        version: '1.0.1',
      };

      execute(link, {
        query: sampleQuery,
        variables,
        context: {
          clientAwareness,
        },
      }).subscribe(
        makeCallback(resolve, reject, () => {
          const [, options] = fetchMock.lastCall()!;
          const { headers } = options as any;
          expect(headers['apollographql-client-name']).toBeDefined();
          expect(headers['apollographql-client-name']).toEqual(
            clientAwareness.name,
          );
          expect(headers['apollographql-client-version']).toBeDefined();
          expect(headers['apollographql-client-version']).toEqual(
            clientAwareness.version,
          );
        }),
      );
    });

    itAsync('should not add empty client awareness settings to request headers', (resolve, reject) => {
      const variables = { params: 'stub' };
      const link = createHttpLink({
        uri: '/data',
      });

      const hasOwn = Object.prototype.hasOwnProperty;
      const clientAwareness = {};
      execute(link, {
        query: sampleQuery,
        variables,
        context: {
          clientAwareness,
        },
      }).subscribe(
        makeCallback(resolve, reject, () => {
          const [, options] = fetchMock.lastCall()!;
          const { headers } = options as any;
          expect(hasOwn.call(headers, 'apollographql-client-name')).toBe(false);
          expect(hasOwn.call(headers, 'apollographql-client-version')).toBe(
            false,
          );
        }),
      );
    });

    itAsync("throws for GET if the variables can't be stringified", (resolve, reject) => {
      const link = createHttpLink({
        uri: '/data',
        useGETForQueries: true,
        includeUnusedVariables: true,
      });

      let b;
      const a: any = { b };
      b = { a };
      a.b = b;
      const variables = {
        a,
        b,
      };
      execute(link, { query: sampleQuery, variables }).subscribe(
        result => {
          reject('next should have been thrown from the link');
        },
        makeCallback(resolve, reject, (e: ClientParseError) => {
          expect(e.message).toMatch(/Variables map is not serializable/);
          expect(e.parseError.message).toMatch(
            /Converting circular structure to JSON/,
          );
        }),
      );
    });

    itAsync("throws for GET if the extensions can't be stringified", (resolve, reject) => {
      const link = createHttpLink({
        uri: '/data',
        useGETForQueries: true,
        includeExtensions: true,
      });

      let b;
      const a: any = { b };
      b = { a };
      a.b = b;
      const extensions = {
        a,
        b,
      };
      execute(link, { query: sampleQuery, extensions }).subscribe(
        result => {
          reject('next should have been thrown from the link');
        },
        makeCallback(resolve, reject, (e: ClientParseError) => {
          expect(e.message).toMatch(/Extensions map is not serializable/);
          expect(e.parseError.message).toMatch(
            /Converting circular structure to JSON/,
          );
        }),
      );
    });

    it('raises warning if called with concat', () => {
      const link = createHttpLink();
      const _warn = console.warn;
      console.warn = (warning: any) => expect(warning['message']).toBeDefined();
      expect(link.concat((operation, forward) => forward(operation))).toEqual(
        link,
      );
      console.warn = _warn;
    });

    it('does not need any constructor arguments', () => {
      expect(() => createHttpLink()).not.toThrow();
    });

    itAsync('calls next and then complete', (resolve, reject) => {
      const next = jest.fn();
      const link = createHttpLink({ uri: 'data' });
      const observable = execute(link, {
        query: sampleQuery,
      });
      observable.subscribe({
        next,
        error: error => reject(error),
        complete: makeCallback(resolve, reject, () => {
          expect(next).toHaveBeenCalledTimes(1);
        }),
      });
    });

    itAsync('calls error when fetch fails', (resolve, reject) => {
      const link = createHttpLink({ uri: 'error' });
      const observable = execute(link, {
        query: sampleQuery,
      });
      observable.subscribe(
        result => reject('next should not have been called'),
        makeCallback(resolve, reject, (error: TypeError) => {
          expect(error).toEqual(mockError.throws);
        }),
        () => reject('complete should not have been called'),
      );
    });

    itAsync('calls error when fetch fails', (resolve, reject) => {
      const link = createHttpLink({ uri: 'error' });
      const observable = execute(link, {
        query: sampleMutation,
      });
      observable.subscribe(
        result => reject('next should not have been called'),
        makeCallback(resolve, reject, (error: TypeError) => {
          expect(error).toEqual(mockError.throws);
        }),
        () => reject('complete should not have been called'),
      );
    });

    itAsync('unsubscribes without calling subscriber', (resolve, reject) => {
      const link = createHttpLink({ uri: 'data' });
      const observable = execute(link, {
        query: sampleQuery,
      });
      const subscription = observable.subscribe(
        result => reject('next should not have been called'),
        error => reject(error),
        () => reject('complete should not have been called'),
      );
      subscription.unsubscribe();
      expect(subscription.closed).toBe(true);
      setTimeout(resolve, 50);
    });

    const verifyRequest = (
      link: ApolloLink,
      resolve: () => void,
      includeExtensions: boolean,
      reject: (error: any) => any,
    ) => {
      const next = jest.fn();
      const context = { info: 'stub' };
      const variables = { params: 'stub' };

      const observable = execute(link, {
        query: sampleMutation,
        context,
        variables,
      });
      observable.subscribe({
        next,
        error: error => reject(error),
        complete: () => {
          try {
            let body = convertBatchedBody(fetchMock.lastCall()![1]!.body);
            expect(body.query).toBe(print(sampleMutation));
            expect(body.variables).toEqual({});
            expect(body.context).not.toBeDefined();
            if (includeExtensions) {
              expect(body.extensions).toBeDefined();
            } else {
              expect(body.extensions).not.toBeDefined();
            }
            expect(next).toHaveBeenCalledTimes(1);

            resolve();
          } catch (e) {
            reject(e);
          }
        },
      });
    };

    itAsync('passes all arguments to multiple fetch body including extensions', (resolve, reject) => {
      const link = createHttpLink({ uri: 'data', includeExtensions: true });
      verifyRequest(
        link,
        () => verifyRequest(link, resolve, true, reject),
        true,
        reject,
      );
    });

    itAsync('passes all arguments to multiple fetch body excluding extensions', (resolve, reject) => {
      const link = createHttpLink({ uri: 'data' });
      verifyRequest(
        link,
        () => verifyRequest(link, resolve, false, reject),
        false,
        reject,
      );
    });

    itAsync('calls multiple subscribers', (resolve, reject) => {
      const link = createHttpLink({ uri: 'data' });
      const context = { info: 'stub' };
      const variables = { params: 'stub' };

      const observable = execute(link, {
        query: sampleMutation,
        context,
        variables,
      });
      observable.subscribe(subscriber);
      observable.subscribe(subscriber);

      setTimeout(() => {
        expect(subscriber.next).toHaveBeenCalledTimes(2);
        expect(subscriber.complete).toHaveBeenCalledTimes(2);
        expect(subscriber.error).not.toHaveBeenCalled();
        resolve();
      }, 50);
    });

    itAsync('calls remaining subscribers after unsubscribe', (resolve, reject) => {
      const link = createHttpLink({ uri: 'data' });
      const context = { info: 'stub' };
      const variables = { params: 'stub' };

      const observable = execute(link, {
        query: sampleMutation,
        context,
        variables,
      });

      observable.subscribe(subscriber);

      setTimeout(() => {
        const subscription = observable.subscribe(subscriber);
        subscription.unsubscribe();
      }, 10);

      setTimeout(
        makeCallback(resolve, reject, () => {
          expect(subscriber.next).toHaveBeenCalledTimes(1);
          expect(subscriber.complete).toHaveBeenCalledTimes(1);
          expect(subscriber.error).not.toHaveBeenCalled();
          resolve();
        }),
        50,
      );
    });

    itAsync('allows for dynamic endpoint setting', (resolve, reject) => {
      const variables = { params: 'stub' };
      const link = createHttpLink({ uri: 'data' });

      execute(link, {
        query: sampleQuery,
        variables,
        context: { uri: 'data2' },
      }).subscribe(result => {
        expect(result).toEqual(data2);
        resolve();
      });
    });

    itAsync('adds headers to the request from the context', (resolve, reject) => {
      const variables = { params: 'stub' };
      const middleware = new ApolloLink((operation, forward) => {
        operation.setContext({
          headers: { authorization: '1234' },
        });
        return forward(operation).map(result => {
          const { headers } = operation.getContext();
          try {
            expect(headers).toBeDefined();
          } catch (e) {
            reject(e);
          }
          return result;
        });
      });
      const link = middleware.concat(createHttpLink({ uri: 'data' }));

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          const headers = fetchMock.lastCall()![1]!.headers as any;
          expect(headers.authorization).toBe('1234');
          expect(headers['content-type']).toBe('application/json');
          expect(headers.accept).toBe('*/*');
        }),
      );
    });

    itAsync('adds headers to the request from the setup', (resolve, reject) => {
      const variables = { params: 'stub' };
      const link = createHttpLink({
        uri: 'data',
        headers: { authorization: '1234' },
      });

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          const headers = fetchMock.lastCall()![1]!.headers as any;
          expect(headers.authorization).toBe('1234');
          expect(headers['content-type']).toBe('application/json');
          expect(headers.accept).toBe('*/*');
        }),
      );
    });

    itAsync('prioritizes context headers over setup headers', (resolve, reject) => {
      const variables = { params: 'stub' };
      const middleware = new ApolloLink((operation, forward) => {
        operation.setContext({
          headers: { authorization: '1234' },
        });
        return forward(operation);
      });
      const link = middleware.concat(
        createHttpLink({ uri: 'data', headers: { authorization: 'no user' } }),
      );

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          const headers = fetchMock.lastCall()![1]!.headers as any;
          expect(headers.authorization).toBe('1234');
          expect(headers['content-type']).toBe('application/json');
          expect(headers.accept).toBe('*/*');
        }),
      );
    });

    itAsync('adds headers to the request from the context on an operation', (resolve, reject) => {
      const variables = { params: 'stub' };
      const link = createHttpLink({ uri: 'data' });

      const context = {
        headers: { authorization: '1234' },
      };
      execute(link, {
        query: sampleQuery,
        variables,
        context,
      }).subscribe(
        makeCallback(resolve, reject, () => {
          const headers = fetchMock.lastCall()![1]!.headers as any;
          expect(headers.authorization).toBe('1234');
          expect(headers['content-type']).toBe('application/json');
          expect(headers.accept).toBe('*/*');
        }),
      );
    });

    itAsync('adds creds to the request from the context', (resolve, reject) => {
      const variables = { params: 'stub' };
      const middleware = new ApolloLink((operation, forward) => {
        operation.setContext({
          credentials: 'same-team-yo',
        });
        return forward(operation);
      });
      const link = middleware.concat(createHttpLink({ uri: 'data' }));

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          const creds = fetchMock.lastCall()![1]!.credentials;
          expect(creds).toBe('same-team-yo');
        }),
      );
    });

    itAsync('adds creds to the request from the setup', (resolve, reject) => {
      const variables = { params: 'stub' };
      const link = createHttpLink({ uri: 'data', credentials: 'same-team-yo' });

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          const creds = fetchMock.lastCall()![1]!.credentials;
          expect(creds).toBe('same-team-yo');
        }),
      );
    });

    itAsync('prioritizes creds from the context over the setup', (resolve, reject) => {
      const variables = { params: 'stub' };
      const middleware = new ApolloLink((operation, forward) => {
        operation.setContext({
          credentials: 'same-team-yo',
        });
        return forward(operation);
      });
      const link = middleware.concat(
        createHttpLink({ uri: 'data', credentials: 'error' }),
      );

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          const creds = fetchMock.lastCall()![1]!.credentials;
          expect(creds).toBe('same-team-yo');
        }),
      );
    });

    itAsync('adds uri to the request from the context', (resolve, reject) => {
      const variables = { params: 'stub' };
      const middleware = new ApolloLink((operation, forward) => {
        operation.setContext({
          uri: 'data',
        });
        return forward(operation);
      });
      const link = middleware.concat(createHttpLink());

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          const uri = fetchMock.lastUrl();
          expect(uri).toBe('/data');
        }),
      );
    });

    itAsync('adds uri to the request from the setup', (resolve, reject) => {
      const variables = { params: 'stub' };
      const link = createHttpLink({ uri: 'data' });

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          const uri = fetchMock.lastUrl();
          expect(uri).toBe('/data');
        }),
      );
    });

    itAsync('prioritizes context uri over setup uri', (resolve, reject) => {
      const variables = { params: 'stub' };
      const middleware = new ApolloLink((operation, forward) => {
        operation.setContext({
          uri: 'apollo',
        });
        return forward(operation);
      });
      const link = middleware.concat(
        createHttpLink({ uri: 'data', credentials: 'error' }),
      );

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          const uri = fetchMock.lastUrl();

          expect(uri).toBe('/apollo');
        }),
      );
    });

    itAsync('allows uri to be a function', (resolve, reject) => {
      const variables = { params: 'stub' };
      const customFetch: WindowOrWorkerGlobalScope['fetch'] = (uri, options) => {
        const { operationName } = convertBatchedBody(options!.body);
        try {
          expect(operationName).toBe('SampleQuery');
        } catch (e) {
          reject(e);
        }
        return fetch('dataFunc', options);
      };

      const link = createHttpLink({ fetch: customFetch });

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          expect(fetchMock.lastUrl()).toBe('/dataFunc');
        }),
      );
    });

    itAsync('adds fetchOptions to the request from the setup', (resolve, reject) => {
      const variables = { params: 'stub' };
      const link = createHttpLink({
        uri: 'data',
        fetchOptions: { someOption: 'foo', mode: 'no-cors' },
      });

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          const { someOption, mode, headers } = fetchMock.lastCall()![1] as any;
          expect(someOption).toBe('foo');
          expect(mode).toBe('no-cors');
          expect(headers['content-type']).toBe('application/json');
        }),
      );
    });

    itAsync('adds fetchOptions to the request from the context', (resolve, reject) => {
      const variables = { params: 'stub' };
      const middleware = new ApolloLink((operation, forward) => {
        operation.setContext({
          fetchOptions: {
            someOption: 'foo',
          },
        });
        return forward(operation);
      });
      const link = middleware.concat(createHttpLink({ uri: 'data' }));

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          const { someOption } = fetchMock.lastCall()![1] as any;
          expect(someOption).toBe('foo');
          resolve();
        }),
      );
    });

    itAsync('uses the latest window.fetch function if options.fetch not configured', (resolve, reject) => {
      const httpLink = createHttpLink({ uri: 'data' });

      const fetch = window.fetch;
      expect(typeof fetch).toBe('function');

      const fetchSpy = jest.spyOn(window, 'fetch');
      fetchSpy.mockImplementation(() => Promise.resolve<Response>({
        text() {
          return Promise.resolve(JSON.stringify({
            data: { hello: "from spy" },
          }));
        },
      } as Response));

      const spyFn = window.fetch;
      expect(spyFn).not.toBe(fetch);

      subscriptions.add(execute(httpLink, {
        query: sampleQuery,
      }).subscribe({
        error: reject,

        next(result) {
          expect(fetchSpy).toHaveBeenCalledTimes(1);
          expect(result).toEqual({
            data: { hello: "from spy" },
          });

          fetchSpy.mockRestore();
          expect(window.fetch).toBe(fetch);

          subscriptions.add(execute(httpLink, {
            query: sampleQuery,
          }).subscribe({
            error: reject,
            next(result) {
              expect(result).toEqual({
                data: { hello: "world" },
              });
              resolve();
            },
          }));
        },
      }));
    });

    itAsync('uses the print option function when defined', (resolve, reject) => {
      const customPrinter = jest.fn(
        (ast: ASTNode, originalPrint: typeof print) => {
          return stripIgnoredCharacters(originalPrint(ast));
        }
      );

      const httpLink = createHttpLink({ uri: 'data', print: customPrinter });

      execute(httpLink, {
        query: sampleQuery,
        context: {
          fetchOptions: { method: 'GET' },
        },
      }).subscribe(
        makeCallback(resolve, reject, () => {
          expect(customPrinter).toHaveBeenCalledTimes(1);
          const [uri] = fetchMock.lastCall()!;
          expect(uri).toBe(
            '/data?query=query%20SampleQuery%7Bstub%7Bid%7D%7D&operationName=SampleQuery&variables=%7B%7D',
          );
        }),
      );
    });

    itAsync('prioritizes context over setup', (resolve, reject) => {
      const variables = { params: 'stub' };
      const middleware = new ApolloLink((operation, forward) => {
        operation.setContext({
          fetchOptions: {
            someOption: 'foo',
          },
        });
        return forward(operation);
      });
      const link = middleware.concat(
        createHttpLink({ uri: 'data', fetchOptions: { someOption: 'bar' } }),
      );

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          const { someOption } = fetchMock.lastCall()![1] as any;
          expect(someOption).toBe('foo');
        }),
      );
    });

    itAsync('allows for not sending the query with the request', (resolve, reject) => {
      const variables = { params: 'stub' };
      const middleware = new ApolloLink((operation, forward) => {
        operation.setContext({
          http: {
            includeQuery: false,
            includeExtensions: true,
          },
        });
        operation.extensions.persistedQuery = { hash: '1234' };
        return forward(operation);
      });
      const link = middleware.concat(createHttpLink({ uri: 'data' }));

      execute(link, { query: sampleQuery, variables }).subscribe(
        makeCallback(resolve, reject, () => {
          let body = convertBatchedBody(fetchMock.lastCall()![1]!.body);

          expect(body.query).not.toBeDefined();
          expect(body.extensions).toEqual({ persistedQuery: { hash: '1234' } });
          resolve();
        }),
      );
    });

    itAsync('sets the raw response on context', (resolve, reject) => {
      const middleware = new ApolloLink((operation, forward) => {
        return new Observable(ob => {
          const op = forward(operation);
          const sub = op.subscribe({
            next: ob.next.bind(ob),
            error: ob.error.bind(ob),
            complete: makeCallback(resolve, reject, () => {
              expect(operation.getContext().response.headers.toBeDefined);
              ob.complete();
            }),
          });

          return () => {
            sub.unsubscribe();
          };
        });
      });

      const link = middleware.concat(createHttpLink({ uri: 'data', fetch }));

      execute(link, { query: sampleQuery }).subscribe(
        result => {
          resolve();
        },
        () => {},
      );
    });
  });

  describe('Dev warnings', () => {
    voidFetchDuringEachTest();

    itAsync('warns if fetch is undeclared', (resolve, reject) => {
      try {
        createHttpLink({ uri: 'data' });
        reject("warning wasn't called");
      } catch (e) {
        makeCallback(resolve, reject, () =>
          expect(e.message).toMatch(/has not been found globally/),
        )();
      }
    });

    itAsync('warns if fetch is undefined', (resolve, reject) => {
      window.fetch = undefined as any;
      try {
        createHttpLink({ uri: 'data' });
        reject("warning wasn't called");
      } catch (e) {
        makeCallback(resolve, reject, () =>
          expect(e.message).toMatch(/has not been found globally/),
        )();
      }
    });

    it('does not warn if fetch is undeclared but a fetch is passed', () => {
      expect(() => {
        createHttpLink({ uri: 'data', fetch: (() => {}) as any });
      }).not.toThrow();
    });
  });

  describe('Error handling', () => {
    let responseBody: any;
    const text = jest.fn(() => {
      const responseBodyText = '{}';
      responseBody = JSON.parse(responseBodyText);
      return Promise.resolve(responseBodyText);
    });
    const textWithData = jest.fn(() => {
      responseBody = {
        data: { stub: { id: 1 } },
        errors: [{ message: 'dangit' }],
      };

      return Promise.resolve(JSON.stringify(responseBody));
    });

    const textWithErrors = jest.fn(() => {
      responseBody = {
        errors: [{ message: 'dangit' }],
      };

      return Promise.resolve(JSON.stringify(responseBody));
    });
    const fetch = jest.fn((uri, options) => {
      return Promise.resolve({ text });
    });
    beforeEach(() => {
      fetch.mockReset();
    });
    itAsync('makes it easy to do stuff on a 401', (resolve, reject) => {
      const middleware = new ApolloLink((operation, forward) => {
        return new Observable(ob => {
          fetch.mockReturnValueOnce(Promise.resolve({ status: 401, text }));
          const op = forward(operation);
          const sub = op.subscribe({
            next: ob.next.bind(ob),
            error: makeCallback(resolve, reject, (e: ServerError) => {
              expect(e.message).toMatch(/Received status code 401/);
              expect(e.statusCode).toEqual(401);
              ob.error(e);
            }),
            complete: ob.complete.bind(ob),
          });

          return () => {
            sub.unsubscribe();
          };
        });
      });

      const link = middleware.concat(createHttpLink({ uri: 'data', fetch: fetch as any }));

      execute(link, { query: sampleQuery }).subscribe(
        result => {
          reject('next should have been thrown from the network');
        },
        () => {},
      );
    });

    itAsync('throws an error if response code is > 300', (resolve, reject) => {
      fetch.mockReturnValueOnce(Promise.resolve({ status: 400, text }));
      const link = createHttpLink({ uri: 'data', fetch: fetch as any });

      execute(link, { query: sampleQuery }).subscribe(
        result => {
          reject('next should have been thrown from the network');
        },
        makeCallback(resolve, reject, (e: ServerError) => {
          expect(e.message).toMatch(/Received status code 400/);
          expect(e.statusCode).toBe(400);
          expect(e.result).toEqual(responseBody);
        }),
      );
    });
    itAsync('throws an error if response code is > 300 and returns data', (resolve, reject) => {
      fetch.mockReturnValueOnce(
        Promise.resolve({ status: 400, text: textWithData }),
      );

      const link = createHttpLink({ uri: 'data', fetch: fetch as any });

      let called = false;

      execute(link, { query: sampleQuery }).subscribe(
        result => {
          called = true;
          expect(result).toEqual(responseBody);
        },
        e => {
          expect(called).toBe(true);
          expect(e.message).toMatch(/Received status code 400/);
          expect(e.statusCode).toBe(400);
          expect(e.result).toEqual(responseBody);
          resolve();
        },
      );
    });
    itAsync('throws an error if only errors are returned', (resolve, reject) => {
      fetch.mockReturnValueOnce(
        Promise.resolve({ status: 400, text: textWithErrors }),
      );

      const link = createHttpLink({ uri: 'data', fetch: fetch as any });

      execute(link, { query: sampleQuery }).subscribe(
        result => {
          reject('should not have called result because we have no data');
        },
        e => {
          expect(e.message).toMatch(/Received status code 400/);
          expect(e.statusCode).toBe(400);
          expect(e.result).toEqual(responseBody);
          resolve();
        },
      );
    });
    itAsync('throws an error if empty response from the server ', (resolve, reject) => {
      fetch.mockReturnValueOnce(Promise.resolve({ text }));
      text.mockReturnValueOnce(Promise.resolve('{ "body": "boo" }'));
      const link = createHttpLink({ uri: 'data', fetch: fetch as any });

      execute(link, { query: sampleQuery }).subscribe(
        result => {
          reject('next should have been thrown from the network');
        },
        makeCallback(resolve, reject, (e: Error) => {
          expect(e.message).toMatch(
            /Server response was missing for query 'SampleQuery'/,
          );
        }),
      );
    });
    itAsync("throws if the body can't be stringified", (resolve, reject) => {
      fetch.mockReturnValueOnce(Promise.resolve({ data: {}, text }));
      const link = createHttpLink({
        uri: 'data',
        fetch: fetch as any,
        includeUnusedVariables: true,
      });

      let b;
      const a: any = { b };
      b = { a };
      a.b = b;
      const variables = {
        a,
        b,
      };
      execute(link, { query: sampleQuery, variables }).subscribe(
        result => {
          reject('next should have been thrown from the link');
        },
        makeCallback(resolve, reject, (e: ClientParseError) => {
          expect(e.message).toMatch(/Payload is not serializable/);
          expect(e.parseError.message).toMatch(
            /Converting circular structure to JSON/,
          );
        }),
      );
    });
    itAsync('supports being cancelled and does not throw', (resolve, reject) => {
      let called = false;
      class AbortController {
        signal: {};
        abort = () => {
          called = true;
        };
      }

      (global as any).AbortController = AbortController;

      fetch.mockReturnValueOnce(Promise.resolve({ text }));
      text.mockReturnValueOnce(
        Promise.resolve('{ "data": { "hello": "world" } }'),
      );

      const link = createHttpLink({ uri: 'data', fetch: fetch as any });

      const sub = execute(link, { query: sampleQuery }).subscribe({
        next: result => {
          reject('result should not have been called');
        },
        error: e => {
          reject(e);
        },
        complete: () => {
          reject('complete should not have been called');
        },
      });
      sub.unsubscribe();

      setTimeout(
        makeCallback(resolve, reject, () => {
          delete (global as any).AbortController;
          expect(called).toBe(true);
          fetch.mockReset();
          text.mockReset();
        }),
        150,
      );
    });

    const body = '{';
    const unparsableJson = jest.fn(() => Promise.resolve(body));
    itAsync('throws a Server error if response is > 300 with unparsable json', (resolve, reject) => {
      fetch.mockReturnValueOnce(
        Promise.resolve({ status: 400, text: unparsableJson }),
      );
      const link = createHttpLink({ uri: 'data', fetch: fetch as any });

      execute(link, { query: sampleQuery }).subscribe(
        result => {
          reject('next should have been thrown from the network');
        },
        makeCallback(resolve, reject, (e: ServerParseError) => {
          expect(e.message).toMatch("Response not successful: Received status code 400");
          expect(e.statusCode).toBe(400);
          expect(e.response).toBeDefined();
          expect(e.bodyText).toBe(undefined);
        }),
      );
    });

    itAsync('throws a ServerParse error if response is 200 with unparsable json', (resolve, reject) => {
      fetch.mockReturnValueOnce(
        Promise.resolve({ status: 200, text: unparsableJson }),
      );
      const link = createHttpLink({ uri: 'data', fetch: fetch as any });

      execute(link, { query: sampleQuery }).subscribe(
        result => {
          reject('next should have been thrown from the network');
        },
        makeCallback(resolve, reject, (e: ServerParseError) => {
          expect(e.message).toMatch(/JSON/);
          expect(e.statusCode).toBe(200);
          expect(e.response).toBeDefined();
          expect(e.bodyText).toBe(body);
        }),
      );
    });
  });

  describe('Multipart responses', () => {
    let originalTextDecoder: any;
    beforeAll(() => {
      originalTextDecoder = TextDecoder;
      (globalThis as any).TextDecoder = TextDecoder;
    });

    afterAll(() => {
      globalThis.TextDecoder = originalTextDecoder;
    });

    const body = [
      '---',
      'Content-Type: application/json; charset=utf-8',
      'Content-Length: 43',
      '',
      '{"data":{"stub":{"id":"0"}},"hasNext":true}',
      '---',
      'Content-Type: application/json; charset=utf-8',
      'Content-Length: 58',
      '',
      '{"hasNext":false, "incremental": [{"data":{"name":"stubby"},"path":["stub"],"extensions":{"timestamp":1633038919}}]}',
      '-----',
    ].join("\r\n");

    it('can handle whatwg stream bodies', (done) => {
      const stream = new ReadableStream({
        async start(controller) {
          const lines = body.split("\r\n");
          try {
            for (const line of lines) {
              await new Promise((resolve) => setTimeout(resolve, 10));
              controller.enqueue(line + "\r\n");
            }
          } finally {
            controller.close();
          }
        },
      });

      const fetch = jest.fn(async () => ({
        status: 200,
        body: stream,
        headers: new Headers({ 'content-type': 'multipart/mixed' }),
      }));

      const link = new HttpLink({
        fetch: fetch as any,
      });

      let i = 0;
      execute(link, { query: sampleDeferredQuery }).subscribe(
        result => {
          try {
            if (i === 0) {
              expect(result).toEqual({
                data: {
                  stub: {
                    id: "0",
                  },
                },
                hasNext: true,
              });
            } else if (i === 1) {
              expect(result).toEqual({
                incremental: [{
                  data: {
                    name: 'stubby',
                  },
                  extensions: {
                    timestamp: 1633038919,
                  },
                  path: ['stub'],
                }],
                hasNext: false,
              });
            }

          } catch (err) {
            done(err);
          } finally {
            i++;
          }
        },
        err => {
          done(err);
        },
        () => {
          if (i !== 2) {
            done(new Error("Unexpected end to observable"));
          }

          done();
        },
      );
    });

    it('can handle node stream bodies', (done) => {
      const stream = Readable.from(body.split("\r\n").map((line) => line + "\r\n"));

      const fetch = jest.fn(async () => ({
        status: 200,
        body: stream,
        headers: new Headers({ 'Content-Type': 'multipart/mixed;boundary="-";deferSpec=20220824' }),
      }));
      const link = new HttpLink({
        fetch: fetch as any,
      });

      let i = 0;
      execute(link, { query: sampleDeferredQuery }).subscribe(
        result => {
          try {
            if (i === 0) {
              expect(result).toEqual({
                data: {
                  stub: {
                    id: "0",
                  },
                },
                hasNext: true,
              });
            } else if (i === 1) {
              expect(result).toEqual({
                incremental: [{
                  data: {
                    name: 'stubby',
                  },
                  extensions: {
                    timestamp: 1633038919,
                  },
                  path: ['stub'],
                }],
                hasNext: false,
              });
            }

          } catch (err) {
            done(err);
          } finally {
            i++;
          }
        },
        err => {
          done(err);
        },
        () => {
          if (i !== 2) {
            done(new Error("Unexpected end to observable"));
          }

          done();
        },
      );
    });

    itAsync('sets correct accept header on request with deferred query', (resolve, reject) => {
      const stream = Readable.from(body.split("\r\n").map((line) => line + "\r\n"));
      const fetch = jest.fn(async () => ({
        status: 200,
        body: stream,
        headers: new Headers({ 'Content-Type': 'multipart/mixed' }),
      }));
      const link = new HttpLink({
        fetch: fetch as any,
      });
      execute(link, {
        query: sampleDeferredQuery
      }).subscribe(
        makeCallback(resolve, reject, () => {
          expect(fetch).toHaveBeenCalledWith(
            '/graphql',
            expect.objectContaining({
              headers: {
                "content-type": "application/json",
                accept: "multipart/mixed; deferSpec=20220824, application/json"
              }
            })
          )
        }),
      );
    });

    // ensure that custom directives beginning with '@defer..' do not trigger
    // custom accept header for multipart responses
    itAsync('sets does not set accept header on query with custom directive begging with @defer', (resolve, reject) => {
      const stream = Readable.from(body.split("\r\n").map((line) => line + "\r\n"));
      const fetch = jest.fn(async () => ({
        status: 200,
        body: stream,
        headers: new Headers({ 'Content-Type': 'multipart/mixed' }),
      }));
      const link = new HttpLink({
        fetch: fetch as any,
      });
      execute(link, {
        query: sampleQueryCustomDirective
      }).subscribe(
        makeCallback(resolve, reject, () => {
          expect(fetch).toHaveBeenCalledWith(
            '/graphql',
            expect.objectContaining({
              headers: {
                accept: "*/*",
                "content-type": "application/json",
              }
            })
          )
        }),
      );
    });
  });
});

redpajama_set_name: RedPajamaGithub


text:
package net.stickycode.configured.guice3.finder;

import java.lang.reflect.Type;
import java.util.Set;

import javax.inject.Inject;

import com.google.inject.ConfigurationException;
import com.google.inject.Injector;
import com.google.inject.Key;
import com.google.inject.ProvisionException;
import com.google.inject.TypeLiteral;
import com.google.inject.util.Types;

import net.stickycode.configured.finder.BeanFinder;
import net.stickycode.configured.finder.BeanNotFoundException;
import net.stickycode.stereotype.StickyComponent;

@StickyComponent
public class GuiceBeanFinder
    implements BeanFinder {

  @Inject
  Injector injector;

  @Override
  public <T> T find(Class<T> type) throws BeanNotFoundException {
    try {
      return injector.getInstance(type);
    }
    catch (ProvisionException e) {
      throw new BeanNotFoundException(e, type);
    }
    catch (ConfigurationException e) {
      try {
        Set<T> beans = injector.getInstance(Key.get(setOf(TypeLiteral.get(type))));
        throw new BeanNotFoundException(type, beans);
      }
      catch (ConfigurationException e2) {
        throw new BeanNotFoundException(e, type);
      }
    }
  }

  @SuppressWarnings("unchecked")
  // wrapping a T in a Set safely returns a Set<T>
  static <T> TypeLiteral<Set<T>> setOf(TypeLiteral<T> elementType) {
    Type type = Types.setOf(elementType.getType());
    return (TypeLiteral<Set<T>>) TypeLiteral.get(type);
  }
}


text:
{-# LANGUAGE CPP #-}
{-# LANGUAGE DeriveFunctor #-}
{-# LANGUAGE DeriveFoldable #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE DeriveTraversable #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE DeriveDataTypeable #-}
{-# LANGUAGE UndecidableInstances #-}
{-# LANGUAGE MultiParamTypeClasses #-}
--------------------------------------------------------------------
-- |
-- Copyright :  (c) Edward Kmett and Dan Doel 2012-2013
-- License   :  BSD3
-- Maintainer:  Edward Kmett <ekmett@gmail.com>
-- Stability :  experimental
-- Portability: non-portable
--
-- These combinators can be used to retain sharing information.
--------------------------------------------------------------------
module Ermine.Unification.Sharing
  ( runSharing
  , withSharing
  , sharing
  , SharingT(..)
  , Shared(..)
  , uncaring
  ) where

import Control.Applicative
import Control.Monad (void)
import Control.Monad.Writer.Class
import Control.Monad.Reader.Class
import Control.Monad.State.Class
import Control.Monad.Trans.Class
import Control.Monad.IO.Class
import Control.Comonad
import Data.Foldable
import Data.Monoid
import Data.Traversable
import Data.Data

data Shared a = Shared !Bool a
  deriving (Eq,Ord,Show,Read,Functor,Foldable,Traversable,Typeable,Data)

instance Comonad Shared where
  extract (Shared _ a) = a
  extend f s@(Shared b _) = Shared b (f s)

-- An efficient strict-in-the-monoid version of WriterT Any@
newtype SharingT m a = SharingT { unsharingT :: m (Shared a) }
#if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ >= 708
  deriving Typeable
#endif

instance Monad m => Functor (SharingT m) where
  fmap f (SharingT m) = SharingT $ do
    Shared p a <- m
    return $! Shared p (f a)
  {-# INLINE fmap #-}

instance Monad m => Applicative (SharingT m) where
  pure a = SharingT (return (Shared False a))
  {-# INLINE pure #-}
  SharingT mf <*> SharingT ma = SharingT $ do
    Shared p f <- mf
    Shared q a <- ma
    return $! Shared (p || q) (f a)
  {-# INLINE (<*>) #-}

instance Monad m => Monad (SharingT m) where
  return a = SharingT (return (Shared False a))
  {-# INLINE return #-}
  SharingT m >>= f = SharingT $ do
    Shared p a <- m
    Shared q b <- unsharingT (f a)
    return $! Shared (p || q) b
  {-# INLINE (>>=) #-}

instance Monad m => MonadWriter Any (SharingT m) where
  tell (Any p) = SharingT $ return $ Shared p ()
  {-# INLINE tell #-}
  listen (SharingT ma) = SharingT $ do
    Shared p a <- ma
    return $! Shared p (a, Any p)
  {-# INLINE listen #-}
  pass (SharingT mapp) = SharingT $ do
    Shared p (a, pp) <- mapp
    return $! Shared (getAny (pp (Any p))) a
  {-# INLINE pass #-}

instance MonadTrans SharingT where
  lift ma = SharingT $ do
    a <- ma
    return $! Shared False a
  {-# INLINE lift #-}

instance MonadIO m => MonadIO (SharingT m) where
  liftIO = lift . liftIO
  {-# INLINE liftIO #-}

instance MonadState s m => MonadState s (SharingT m) where
  get = lift get
  {-# INLINE get #-}
  put = lift . put
  {-# INLINE put #-}

instance MonadReader e m => MonadReader e (SharingT m) where
  ask = lift ask
  {-# INLINE ask #-}
  local f = SharingT . local f . unsharingT
  {-# INLINE local #-}

-- | Run an action, if it returns @'Any' 'True'@ then use its new value, otherwise use the passed in value.
--
-- This can be used to recover sharing during unification when no interesting unification takes place.
--
-- This version discards the 'SharingT' wrapper.
runSharing :: Monad m => a -> SharingT m a -> m a
runSharing a m = do
  Shared modified b <- unsharingT m
  return $! if modified then b else a
{-# INLINE runSharing #-}

withSharing :: Monad m => (a -> SharingT m a) -> a -> m a
withSharing k a = runSharing a (k a)
{-# INLINE withSharing #-}

uncaring :: Functor m => SharingT m a -> m ()
uncaring = void . unsharingT
{-# INLINE uncaring #-}

-- | Run an action, if it returns @'Any' 'True'@ then use its new value, otherwise use the passed in value.
--
-- This can be used to recover sharing during unification when no interesting unification takes place.
--
-- This version retains the current monad wrapper.
sharing :: MonadWriter Any m => a -> m a -> m a
sharing a m = do
  (b, Any modified) <- listen m
  return $! if modified then b else a
{-# INLINE sharing #-}


text:
using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
using EasyNetQ.Consumer;
using EasyNetQ.Internals;
using EasyNetQ.Persistent;
using EasyNetQ.Producer;

namespace EasyNetQ.ChannelDispatcher;

/// <summary>
///     Invokes client commands using single channel
/// </summary>
public sealed class SinglePersistentChannelDispatcher : IPersistentChannelDispatcher
{
    private readonly ConcurrentDictionary<PersistentChannelDispatchOptions, IPersistentChannel> channelPerOptions;
    private readonly Func<PersistentChannelDispatchOptions, IPersistentChannel> createChannelFactory;

    /// <summary>
    /// Creates a dispatcher
    /// </summary>
    public SinglePersistentChannelDispatcher(
        IProducerConnection producerConnection,
        IConsumerConnection consumerConnection,
        IPersistentChannelFactory channelFactory
    )
    {
        Preconditions.CheckNotNull(producerConnection, nameof(producerConnection));
        Preconditions.CheckNotNull(consumerConnection, nameof(consumerConnection));
        Preconditions.CheckNotNull(channelFactory, nameof(channelFactory));

        channelPerOptions = new ConcurrentDictionary<PersistentChannelDispatchOptions, IPersistentChannel>();
        createChannelFactory = o =>
        {
            var options = new PersistentChannelOptions(o.PublisherConfirms);
            return o.ConnectionType switch
            {
                PersistentConnectionType.Producer => channelFactory.CreatePersistentChannel(
                    producerConnection, options
                ),
                PersistentConnectionType.Consumer => channelFactory.CreatePersistentChannel(
                    consumerConnection, options
                ),
                _ => throw new ArgumentOutOfRangeException()
            };
        };
    }

    /// <inheritdoc />
    public Task<TResult> InvokeAsync<TResult, TChannelAction>(
        TChannelAction channelAction,
        PersistentChannelDispatchOptions options,
        CancellationToken cancellationToken = default
    ) where TChannelAction : struct, IPersistentChannelAction<TResult>
    {
        // TODO createChannelFactory could be called multiple time, fix it
        var channel = channelPerOptions.GetOrAdd(options, createChannelFactory);
        return channel.InvokeChannelActionAsync<TResult, TChannelAction>(channelAction, cancellationToken);
    }

    /// <inheritdoc />
    public void Dispose() => channelPerOptions.ClearAndDispose();
}


text:
var utils = require('./transpiled/utils').default;
module.exports = utils


text:
using System;

namespace CKS.Contracts
{
    public interface ILogger
    {
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Naming", "CA1716:IdentifiersShouldNotMatchKeywords", MessageId = "Error")]
        void Error(string message);
        void ErrorException(string message, Exception exception);
        void Info(string message);
        void InfoException(string message, Exception exception);
    }
}