/* This is the thread function that handles the calling of callback functions if set */
thread_return_type WINAPI MQTTClient_run(void* n)
{
	long timeout = 10L; /* first time in we have a small timeout.  Gets things started more quickly */

	FUNC_ENTRY;
	running = 1;
	run_id = Thread_getid();

	Thread_lock_mutex(mqttclient_mutex);
	while (!tostop)
	{
		int rc = SOCKET_ERROR;
		int sock = -1;
		MQTTClients* m = NULL;
		MQTTPacket* pack = NULL;

		Thread_unlock_mutex(mqttclient_mutex);
		pack = MQTTClient_cycle(&sock, timeout, &rc);
		Thread_lock_mutex(mqttclient_mutex);
		if (tostop)
			break;
		timeout = 1000L;

		/* find client corresponding to socket */
		if (ListFindItem(handles, &sock, clientSockCompare) == NULL)
		{
			/* assert: should not happen */
			continue;
		}
		m = (MQTTClient)(handles->current->content);
		if (m == NULL)
		{
			/* assert: should not happen */
			continue;
		}
		if (rc == SOCKET_ERROR)
		{
			Thread_unlock_mutex(mqttclient_mutex);
			MQTTClient_disconnect_internal(m, 0);
			Thread_lock_mutex(mqttclient_mutex);
		}
		else
		{
			if (m->c->messageQueue->count > 0)
			{
				qEntry* qe = (qEntry*)(m->c->messageQueue->first->content);
				int topicLen = qe->topicLen;

				if (strlen(qe->topicName) == topicLen)
					topicLen = 0;

				Log(TRACE_MIN, -1, "Calling messageArrived for client %s, queue depth %d",
					m->c->clientID, m->c->messageQueue->count);
				Thread_unlock_mutex(mqttclient_mutex);
				rc = (*(m->ma))(m->context, qe->topicName, topicLen, qe->msg);
				Thread_lock_mutex(mqttclient_mutex);
				/* if 0 (false) is returned by the callback then it failed, so we don't remove the message from
				 * the queue, and it will be retried later.  If 1 is returned then the message data may have been freed,
				 * so we must be careful how we use it.
				 */
				if (rc)
					ListRemove(m->c->messageQueue, qe);
				else
					Log(TRACE_MIN, -1, "False returned from messageArrived for client %s, message remains on queue",
						m->c->clientID);
			}
			if (pack)
			{
				if (pack->header.bits.type == CONNACK)
				{
					Log(TRACE_MIN, -1, "Posting connack semaphore for client %s", m->c->clientID);
					m->pack = pack;
					Thread_post_sem(m->connack_sem);
				}
				else if (pack->header.bits.type == SUBACK)
				{
					Log(TRACE_MIN, -1, "Posting suback semaphore for client %s", m->c->clientID);
					m->pack = pack;
					Thread_post_sem(m->suback_sem);
				}
				else if (pack->header.bits.type == UNSUBACK)
				{
					Log(TRACE_MIN, -1, "Posting unsuback semaphore for client %s", m->c->clientID);
					m->pack = pack;
					Thread_post_sem(m->unsuback_sem);
				}
			}
			else if (m->c->connect_state == 1 && !Thread_check_sem(m->connect_sem))
			{
				int error;
				socklen_t len = sizeof(error);

				if ((m->rc = getsockopt(m->c->net.socket, SOL_SOCKET, SO_ERROR, (char*)&error, &len)) == 0)
					m->rc = error;
				Log(TRACE_MIN, -1, "Posting connect semaphore for client %s rc %d", m->c->clientID, m->rc);
				Thread_post_sem(m->connect_sem);
			}
#if defined(OPENSSL)
			else if (m->c->connect_state == 2 && !Thread_check_sem(m->connect_sem))
//.........这里部分代码省略.........

bool KeyMap::InitializeKeyMap(const char *inifile, const char *tablefile)
{
	AutoTable kmtable(tablefile);

	if (!kmtable) {
		return false;
	}

	char tINIkeymap[_MAX_PATH];
	PathJoin( tINIkeymap, core->GamePath, inifile, NULL );
	FileStream* config = FileStream::OpenFile( tINIkeymap );

	if (config == NULL) {
		Log(WARNING, "KeyMap", "There is no '%s' file...", inifile);
		return false;
	}
	char name[KEYLENGTH+1], value[_MAX_PATH + 3];
	while (config->Remains()) {
		char line[_MAX_PATH];

		if (config->ReadLine(line, _MAX_PATH) == -1)
			break;

		if ((line[0] == '#') ||
			( line[0] == '[' ) ||
			( line[0] == '\r' ) ||
			( line[0] == '\n' ) ||
			( line[0] == ';' )) {
			continue;
		}

		name[0] = 0;
		value[0] = 0;

		//ignore possible space after the =, sadly we cannot do the same with
		//spaces before it
		if (sscanf( line, "%[^=]= %[^\r\n]", name, value )!=2)
			continue;

		strnlwrcpy(name,name,KEYLENGTH);
		//remove trailing spaces (bg1 ini file contains them)
		char *nameend = name + strlen( name ) - 1;
		while (nameend >= name && strchr( " \t\r\n", *nameend )) {
			*nameend-- = '\0';
		}

		//change internal spaces to underscore
		for(int c=0;c<KEYLENGTH;c++) if (name[c]==' ') name[c]='_';

		int l = strlen(value);
		Function *fun;
		void *tmp;

		if (l<0 || l>1 || keymap.Lookup(value, tmp) ) {
			print("Ignoring key %s", value);
			continue;
		}

		const char *module;
		const char *function;
		const char *group;

		if (kmtable->GetRowIndex(name)>=0 ) {
			module = kmtable->QueryField(name, "MODULE");
			function = kmtable->QueryField(name, "FUNCTION");
			group = kmtable->QueryField(name, "GROUP");
		} else {
			module = kmtable->QueryField("Default","MODULE");
			function = kmtable->QueryField("Default","FUNCTION");
			group = kmtable->QueryField("Default","GROUP");
			print("Adding key %s with function %s::%s", value, module, function);
		}
		fun = new Function(module, function, atoi(group));
		keymap.SetAt(value, fun);
	}
	delete config;
	return true;
}

//todo: this function is an abomination, this is just disgusting.  fix it.
//...seriously, this is really, really horrible.  I mean this is amazingly bad.
void OBS::MainCaptureLoop()
{
    int curRenderTarget = 0, curYUVTexture = 0, curCopyTexture = 0;
    int copyWait = NUM_RENDER_BUFFERS-1;

    bSentHeaders = false;
    bFirstAudioPacket = true;

    bool bLogLongFramesProfile = GlobalConfig->GetInt(TEXT("General"), TEXT("LogLongFramesProfile"), LOGLONGFRAMESDEFAULT) != 0;
    float logLongFramesProfilePercentage = GlobalConfig->GetFloat(TEXT("General"), TEXT("LogLongFramesProfilePercentage"), 10.f);

    Vect2 baseSize    = Vect2(float(baseCX), float(baseCY));
    Vect2 outputSize  = Vect2(float(outputCX), float(outputCY));
    Vect2 scaleSize   = Vect2(float(scaleCX), float(scaleCY));

    HANDLE hMatrix   = yuvScalePixelShader->GetParameterByName(TEXT("yuvMat"));
    HANDLE hScaleVal = yuvScalePixelShader->GetParameterByName(TEXT("baseDimensionI"));

    //----------------------------------------
    // x264 input buffers

    int curOutBuffer = 0;

    bool bUsingQSV = videoEncoder->isQSV();//GlobalConfig->GetInt(TEXT("Video Encoding"), TEXT("UseQSV")) != 0;
    bUsing444 = false;

    EncoderPicture lastPic;
    EncoderPicture outPics[NUM_OUT_BUFFERS];

    for(int i=0; i<NUM_OUT_BUFFERS; i++)
    {
        if(bUsingQSV)
        {
            outPics[i].mfxOut = new mfxFrameSurface1;
            memset(outPics[i].mfxOut, 0, sizeof(mfxFrameSurface1));
            mfxFrameData& data = outPics[i].mfxOut->Data;
            videoEncoder->RequestBuffers(&data);
        }
        else
        {
            outPics[i].picOut = new x264_picture_t;
            x264_picture_init(outPics[i].picOut);
        }
    }

    if(bUsing444)
    {
        for(int i=0; i<NUM_OUT_BUFFERS; i++)
        {
            outPics[i].picOut->img.i_csp   = X264_CSP_BGRA; //although the x264 input says BGR, x264 actually will expect packed UYV
            outPics[i].picOut->img.i_plane = 1;
        }
    }
    else
    {
        if(!bUsingQSV)
            for(int i=0; i<NUM_OUT_BUFFERS; i++)
                x264_picture_alloc(outPics[i].picOut, X264_CSP_NV12, outputCX, outputCY);
    }

    int bCongestionControl = AppConfig->GetInt (TEXT("Video Encoding"), TEXT("CongestionControl"), 0);
    bool bDynamicBitrateSupported = App->GetVideoEncoder()->DynamicBitrateSupported();
    int defaultBitRate = AppConfig->GetInt(TEXT("Video Encoding"), TEXT("MaxBitrate"), 1000);
    int currentBitRate = defaultBitRate;
    QWORD lastAdjustmentTime = 0;
    UINT adjustmentStreamId = 0;

    //std::unique_ptr<ProfilerNode> encodeThreadProfiler;

    //----------------------------------------
    // time/timestamp stuff

    bool bWasLaggedFrame = false;

    totalStreamTime = 0;
    lastAudioTimestamp = 0;

    //----------------------------------------
    // start audio capture streams

    desktopAudio->StartCapture();
    if(micAudio) micAudio->StartCapture();

    //----------------------------------------
    // status bar/statistics stuff

    DWORD fpsCounter = 0;

    int numLongFrames = 0;
    int numTotalFrames = 0;

    bytesPerSec = 0;
    captureFPS = 0;
    curFramesDropped = 0;
    curStrain = 0.0;
    PostMessage(hwndMain, OBS_UPDATESTATUSBAR, 0, 0);

    QWORD lastBytesSent[3] = {0, 0, 0};
//.........这里部分代码省略.........

int SQLI_compose_static_queries()
{
  int primitives=0, set_primitives=0, set_event_primitives=0, have_flows=0;

  if (config.what_to_count & COUNT_FLOWS || (config.sql_table_version >= 4 &&
                                             config.sql_table_version < SQL_TABLE_VERSION_BGP &&
                                             !config.sql_optimize_clauses)) {
    config.what_to_count |= COUNT_FLOWS;
    have_flows = TRUE;

    if ((config.sql_table_version < 4 || config.sql_table_version >= SQL_TABLE_VERSION_BGP) && !config.sql_optimize_clauses) {
      Log(LOG_ERR, "ERROR ( %s/%s ): The accounting of flows requires SQL table v4. Exiting.\n", config.name, config.type);
      exit_plugin(1);
    }
  }

  /* "INSERT INTO ... VALUES ... " and "... WHERE ..." stuff */
  strncpy(where[primitives].string, " WHERE ", sizeof(where[primitives].string));
  snprintf(insert_clause, sizeof(insert_clause), "INSERT INTO %s (", config.sql_table);
  strncpy(values[primitives].string, " VALUES (", sizeof(values[primitives].string));
  primitives = SQLI_evaluate_history(primitives);
  primitives = sql_evaluate_primitives(primitives);

  strncpy(insert_counters_clause, ", packets, bytes", SPACELEFT(insert_counters_clause));
  if (have_flows) strncat(insert_counters_clause, ", flows", SPACELEFT(insert_counters_clause));
  strncat(insert_counters_clause, ")", SPACELEFT(insert_counters_clause));
  strncpy(insert_nocounters_clause, ")", SPACELEFT(insert_nocounters_clause));

  /* "LOCK ..." stuff */
  if (config.sql_locking_style) Log(LOG_WARNING, "WARN ( %s/%s ): sql_locking_style is not supported. Ignored.\n", config.name, config.type);
  snprintf(lock_clause, sizeof(lock_clause), "BEGIN", config.sql_table);
  strncpy(unlock_clause, "COMMIT", sizeof(unlock_clause));

  /* "UPDATE ... SET ..." stuff */
  snprintf(update_clause, sizeof(update_clause), "UPDATE %s ", config.sql_table);

  set_primitives = sql_compose_static_set(have_flows);
  set_event_primitives = sql_compose_static_set_event();

  if (config.sql_history) {
    if (!config.sql_history_since_epoch) {
      strncpy(set[set_primitives].string, ", ", SPACELEFT(set[set_primitives].string));
      strncat(set[set_primitives].string, "stamp_updated=DATETIME('now', 'localtime')", SPACELEFT(set[set_primitives].string));
      set[set_primitives].type = TIMESTAMP;
      set[set_primitives].handler = count_noop_setclause_handler;
      set_primitives++;

      if (set_event_primitives) strncpy(set_event[set_event_primitives].string, ", ", SPACELEFT(set_event[set_event_primitives].string));
      else strncpy(set_event[set_event_primitives].string, "SET ", SPACELEFT(set_event[set_event_primitives].string));
      strncat(set_event[set_event_primitives].string, "stamp_updated=DATETIME('now', 'localtime')", SPACELEFT(set_event[set_event_primitives].string));
      set_event[set_event_primitives].type = TIMESTAMP;
      set_event[set_event_primitives].handler = count_noop_setclause_event_handler;
      set_event_primitives++;
    }
    else {
      strncpy(set[set_primitives].string, ", ", SPACELEFT(set[set_primitives].string));
      strncat(set[set_primitives].string, "stamp_updated=STRFTIME('%%s', 'now')", SPACELEFT(set[set_primitives].string));
      set[set_primitives].type = TIMESTAMP;
      set[set_primitives].handler = count_noop_setclause_handler;
      set_primitives++;

      if (set_event_primitives) strncpy(set_event[set_event_primitives].string, ", ", SPACELEFT(set_event[set_event_primitives].string));
      else strncpy(set_event[set_event_primitives].string, "SET ", SPACELEFT(set_event[set_event_primitives].string));
      strncat(set_event[set_event_primitives].string, "stamp_updated=STRFTIME('%%s', 'now')", SPACELEFT(set_event[set_event_primitives].string));
      set_event[set_event_primitives].type = TIMESTAMP;
      set_event[set_event_primitives].handler = count_noop_setclause_event_handler;
      set_event_primitives++;
    }
  }

  return primitives;
}

	bool preprocess(const Scene *scene, RenderQueue *queue, const RenderJob *job,
			int sceneResID, int sensorResID, int samplerResID) {
		SamplingIntegrator::preprocess(scene, queue, job, sceneResID, sensorResID, samplerResID);
		/* Create a deterministic sampler for the photon gathering step */
		ref<Scheduler> sched = Scheduler::getInstance();
		ref<Sampler> sampler = static_cast<Sampler *> (PluginManager::getInstance()->
			createObject(MTS_CLASS(Sampler), Properties("halton")));
		/* Create a sampler instance for every core */
		std::vector<SerializableObject *> samplers(sched->getCoreCount());
		for (size_t i=0; i<sched->getCoreCount(); ++i) {
			ref<Sampler> clonedSampler = sampler->clone();
			clonedSampler->incRef();
			samplers[i] = clonedSampler.get();
		}
		int qmcSamplerID = sched->registerMultiResource(samplers);
		for (size_t i=0; i<samplers.size(); ++i)
			samplers[i]->decRef();

		const ref_vector<Medium> &media = scene->getMedia();
		for (ref_vector<Medium>::const_iterator it = media.begin(); it != media.end(); ++it) {
			if (!(*it)->isHomogeneous())
				Log(EError, "Inhomogeneous media are currently not supported by the photon mapper!");
		}

		if (m_globalPhotonMap.get() == NULL && m_globalPhotons > 0) {
			/* Generate the global photon map */
			ref<GatherPhotonProcess> proc = new GatherPhotonProcess(
				GatherPhotonProcess::ESurfacePhotons, m_globalPhotons,
				m_granularity, m_maxDepth-1, m_rrDepth, m_gatherLocally,
				m_autoCancelGathering, job);

			proc->bindResource("scene", sceneResID);
			proc->bindResource("sensor", sensorResID);
			proc->bindResource("sampler", qmcSamplerID);

			m_proc = proc;
			sched->schedule(proc);
			sched->wait(proc);
			m_proc = NULL;

			if (proc->getReturnStatus() != ParallelProcess::ESuccess)
				return false;

			ref<PhotonMap> globalPhotonMap = proc->getPhotonMap();
			if (globalPhotonMap->isFull()) {
				Log(EDebug, "Global photon map full. Shot " SIZE_T_FMT " particles, excess photons due to parallelism: "
					SIZE_T_FMT, proc->getShotParticles(), proc->getExcessPhotons());

				m_globalPhotonMap = globalPhotonMap;
				m_globalPhotonMap->setScaleFactor(1 / (Float) proc->getShotParticles());
				m_globalPhotonMap->build();
				m_globalPhotonMapID = sched->registerResource(m_globalPhotonMap);
			}
		}

		if (m_causticPhotonMap.get() == NULL && m_causticPhotons > 0) {
			/* Generate the caustic photon map */
			ref<GatherPhotonProcess> proc = new GatherPhotonProcess(
				GatherPhotonProcess::ECausticPhotons, m_causticPhotons,
				m_granularity, m_maxDepth-1, m_rrDepth, m_gatherLocally,
				m_autoCancelGathering, job);

			proc->bindResource("scene", sceneResID);
			proc->bindResource("sensor", sensorResID);
			proc->bindResource("sampler", qmcSamplerID);

			m_proc = proc;
			sched->schedule(proc);
			sched->wait(proc);
			m_proc = NULL;

			if (proc->getReturnStatus() != ParallelProcess::ESuccess)
				return false;

			ref<PhotonMap> causticPhotonMap = proc->getPhotonMap();
			if (causticPhotonMap->isFull()) {
				Log(EDebug, "Caustic photon map full. Shot " SIZE_T_FMT " particles, excess photons due to parallelism: "
					SIZE_T_FMT, proc->getShotParticles(), proc->getExcessPhotons());

				m_causticPhotonMap = causticPhotonMap;
				m_causticPhotonMap->setScaleFactor(1 / (Float) proc->getShotParticles());
				m_causticPhotonMap->build();
				m_causticPhotonMapID = sched->registerResource(m_causticPhotonMap);
			}
		}

		size_t volumePhotons = scene->getMedia().size() == 0 ? 0 : m_volumePhotons;
		if (m_volumePhotonMap.get() == NULL && volumePhotons > 0) {
			/* Generate the volume photon map */
			ref<GatherPhotonProcess> proc = new GatherPhotonProcess(
				GatherPhotonProcess::EVolumePhotons, volumePhotons,
				m_granularity, m_maxDepth-1, m_rrDepth, m_gatherLocally,
				m_autoCancelGathering, job);

			proc->bindResource("scene", sceneResID);
			proc->bindResource("sensor", sensorResID);
			proc->bindResource("sampler", qmcSamplerID);

			m_proc = proc;
			sched->schedule(proc);
//.........这里部分代码省略.........

int InitializeOpenCL(void)
{
  if (numDevices != -12345)
    return numDevices;

  numDevices = -1;  /* assume detection failure for now */

  cl_uint devicesDetected = 0;
  cl_uint numPlatforms;
  cl_int status = clGetPlatformIDs(0, NULL, &numPlatforms);
  if (status != CL_SUCCESS)
  {
    Log("Error obtaining number of platforms (clGetPlatformIDs/1)\n");
    ocl_diagnose(status, NULL, NULL);  // decode error code only
  }
  if (status == CL_SUCCESS)
  {
    cl_platform_id *platforms = NULL;
    cl_device_id *devices = NULL;

    if (numPlatforms != 0)
    {
      // Allocate enough space for each platform
      platforms = (cl_platform_id *) malloc(numPlatforms * sizeof(cl_platform_id));

      // Fill in platforms with clGetPlatformIDs()
      status = clGetPlatformIDs(numPlatforms, platforms, NULL);
      if (status != CL_SUCCESS)
      {
        Log("Error obtaining list of platforms (clGetPlatformIDs/2)\n");
        ocl_diagnose(status, NULL, NULL);  // decode error code only
      }
      else
      {
        // Use clGetDeviceIDs() to retrieve the number of devices present
        for (cl_uint plat = 0; plat < numPlatforms; plat++)
        {
          cl_uint devcnt;

          status = clGetDeviceIDs(platforms[plat], CL_DEVICE_TYPE_GPU, 0, NULL, &devcnt);
          if (status == CL_DEVICE_NOT_FOUND)  // Special case. No GPU devices but other may exist
          {
            status = CL_SUCCESS;
            devcnt = 0;
          }
          if (status != CL_SUCCESS)
          {
            Log("Error obtaining number of devices on platform %u (clGetDeviceIDs/1)\n", plat);
            ocl_diagnose(status, NULL, NULL);  // decode error code only
            break;
          }
          devicesDetected += devcnt;
        }
      }
    }

    if (status == CL_SUCCESS && devicesDetected != 0)
    {
      // Allocate enough space for each device
      devices = (cl_device_id*) malloc(devicesDetected * sizeof(cl_device_id));

      // Allocate and zero space for ocl_context
      ocl_context = (ocl_context_t*) calloc(devicesDetected, sizeof(ocl_context_t));

      // Fill in devices with clGetDeviceIDs()
      cl_uint offset = 0;
      for (cl_uint plat = 0; plat < numPlatforms; plat++)
      {
        cl_uint devcnt;

        if (offset >= devicesDetected)   /* Avoid call with bufferSize=0 for last platform without GPU devices */
          break;

        status = clGetDeviceIDs(platforms[plat], CL_DEVICE_TYPE_GPU, devicesDetected - offset, devices + offset, &devcnt);
        if (status == CL_DEVICE_NOT_FOUND)  // Special case. No GPU devices but other may exist
        {
          status = CL_SUCCESS;
          devcnt = 0;
        }
        if (status != CL_SUCCESS)
        {
          Log("Error obtaining list of devices on platform %u (clGetDeviceIDs/2)\n", plat);
          ocl_diagnose(status, NULL, NULL);  // decode error code only
          break;
        }

        // Fill non-zero context fields for each device
        for (cl_uint u = 0; u < devcnt; u++, offset++)
        {
          ocl_context_t *cont = &ocl_context[offset];

          /* Assume it working for now */
          cont->active            = true;
          cont->coreID            = CORE_NONE;
          cont->platformID        = platforms[plat];
          cont->deviceID          = devices[offset];
          cont->firstOnPlatform   = (u == 0);
          cont->clientDeviceNo    = offset;
          cont->runSize           = 65536;
          cont->runSizeMultiplier = 64;
//.........这里部分代码省略.........

/* Functions */
void sqlite3_plugin(int pipe_fd, struct configuration *cfgptr, void *ptr) 
{
  struct pkt_data *data;
  struct ports_table pt;
  struct pollfd pfd;
  struct insert_data idata;
  time_t refresh_deadline;
  int timeout, refresh_timeout, amqp_timeout;
  int ret, num;
  struct ring *rg = &((struct channels_list_entry *)ptr)->rg;
  struct ch_status *status = ((struct channels_list_entry *)ptr)->status;
  struct plugins_list_entry *plugin_data = ((struct channels_list_entry *)ptr)->plugin;
  int datasize = ((struct channels_list_entry *)ptr)->datasize;
  u_int32_t bufsz = ((struct channels_list_entry *)ptr)->bufsize;
  pid_t core_pid = ((struct channels_list_entry *)ptr)->core_pid;
  struct networks_file_data nfd;
  char *dataptr;

  unsigned char *rgptr;
  int pollagain = TRUE;
  u_int32_t seq = 1, rg_err_count = 0; 

  struct extra_primitives extras;
  struct primitives_ptrs prim_ptrs;

#ifdef WITH_RABBITMQ
  struct p_amqp_host *amqp_host = &((struct channels_list_entry *)ptr)->amqp_host;
#endif

  memcpy(&config, cfgptr, sizeof(struct configuration));
  memcpy(&extras, &((struct channels_list_entry *)ptr)->extras, sizeof(struct extra_primitives));
  recollect_pipe_memory(ptr);
  pm_setproctitle("%s [%s]", "SQLite3 Plugin", config.name);

  memset(&idata, 0, sizeof(idata));
  if (config.pidfile) write_pid_file_plugin(config.pidfile, config.type, config.name);
  if (config.logfile) {
    fclose(config.logfile_fd);
    config.logfile_fd = open_output_file(config.logfile, "a", FALSE);
  }

  sql_set_signals();
  sql_init_default_values(&extras);
  SQLI_init_default_values(&idata);
  SQLI_set_callbacks(&sqlfunc_cbr);
  sql_set_insert_func();

  /* some LOCAL initialization AFTER setting some default values */
  reload_map = FALSE;
  idata.now = time(NULL);
  refresh_deadline = idata.now;
  idata.cfg = &config;

  sql_init_maps(&extras, &prim_ptrs, &nt, &nc, &pt);
  sql_init_global_buffers();
  sql_init_historical_acct(idata.now, &idata);
  sql_init_triggers(idata.now, &idata);
  sql_init_refresh_deadline(&refresh_deadline);

  if (config.pipe_amqp) {
    plugin_pipe_amqp_compile_check();
#ifdef WITH_RABBITMQ
    pipe_fd = plugin_pipe_amqp_connect_to_consume(amqp_host, plugin_data);
    amqp_timeout = plugin_pipe_set_retry_timeout(&amqp_host->btimers, pipe_fd);
#endif
  }
  else setnonblocking(pipe_fd);

  /* setting number of entries in _protocols structure */
  while (_protocols[protocols_number].number != -1) protocols_number++;

  /* building up static SQL clauses */
  idata.num_primitives = SQLI_compose_static_queries();
  glob_num_primitives = idata.num_primitives; 

  /* setting up environment variables */
  SQL_SetENV();

  sql_link_backend_descriptors(&bed, &p, &b);

  /* plugin main loop */
  for(;;) {
    poll_again:
    status->wakeup = TRUE;
    calc_refresh_timeout(refresh_deadline, idata.now, &refresh_timeout);

    pfd.fd = pipe_fd;
    pfd.events = POLLIN;
    timeout = MIN(refresh_timeout, (amqp_timeout ? amqp_timeout : INT_MAX));
    ret = poll(&pfd, (pfd.fd == ERR ? 0 : 1), timeout);

    if (ret <= 0) {
      if (getppid() == 1) {
        Log(LOG_ERR, "ERROR ( %s/%s ): Core process *seems* gone. Exiting.\n", config.name, config.type);
        exit_plugin(1);
      }

      if (ret < 0) goto poll_again;
    }
//.........这里部分代码省略.........

//.........这里部分代码省略.........
                            if (!Render(out, start, input, hash_stack, delim_start, delim_start_len, delim_end, delim_end_len,
                                        skip, section, &cur_section_end))
                            {
                                free(section);
                                return false;
                            }
                            free(section);
                            input = cur_section_end;
                        }
                        continue;

                    default:
                        assert(false);
                        return false;
                    }
                    break;

                case JSON_ELEMENT_TYPE_CONTAINER:
                    switch (JsonGetContrainerType(var))
                    {
                    case JSON_CONTAINER_TYPE_OBJECT:
                        {
                            const char *cur_section_end = NULL;
                            if (!Render(out, start, input,
                                        hash_stack,
                                        delim_start, delim_start_len, delim_end, delim_end_len,
                                        skip_content || tag.type == TAG_TYPE_INVERTED, section, &cur_section_end))
                            {
                                free(section);
                                return false;
                            }
                            free(section);
                            input = cur_section_end;
                        }
                        break;

                    case JSON_CONTAINER_TYPE_ARRAY:
                        if (JsonLength(var) > 0)
                        {
                            const char *cur_section_end = NULL;
                            for (size_t i = 0; i < JsonLength(var); i++)
                            {
                                JsonElement *child_hash = JsonAt(var, i);
                                SeqAppend(hash_stack, child_hash);


                                if (!Render(out, start, input,
                                            hash_stack,
                                            delim_start, delim_start_len, delim_end, delim_end_len,
                                            skip_content || tag.type == TAG_TYPE_INVERTED, section, &cur_section_end))
                                {
                                    free(section);
                                    return false;
                                }
                            }
                            input = cur_section_end;
                            free(section);
                        }
                        else
                        {
                            const char *cur_section_end = NULL;
                            if (!Render(out, start, input, hash_stack, delim_start, delim_start_len, delim_end, delim_end_len,
                                        tag.type != TAG_TYPE_INVERTED, section, &cur_section_end))
                            {
                                free(section);
                                return false;
                            }
                            free(section);
                            input = cur_section_end;
                        }
                        break;
                    }
                    break;
                }
            }
            continue;
        case TAG_TYPE_SECTION_END:
            if (!section)
            {
                char *varname = xstrndup(tag.content, tag.content_len);
                Log(LOG_LEVEL_WARNING, "Unknown section close in mustache template '%s'", varname);
                free(varname);
                return false;
            }
            else
            {
                SeqRemove(hash_stack, SeqLength(hash_stack) - 1);
                *section_end = input;
                return true;
            }
            break;

        default:
            assert(false);
            return false;
        }
    }

    assert(false);
}

/* Functions */
void amqp_plugin(int pipe_fd, struct configuration *cfgptr, void *ptr) 
{
  struct pkt_data *data;
  struct ports_table pt;
  unsigned char *pipebuf;
  struct pollfd pfd;
  struct insert_data idata;
  time_t t;
  int timeout, refresh_timeout, amqp_timeout, ret, num; 
  struct ring *rg = &((struct channels_list_entry *)ptr)->rg;
  struct ch_status *status = ((struct channels_list_entry *)ptr)->status;
  struct plugins_list_entry *plugin_data = ((struct channels_list_entry *)ptr)->plugin;
  int datasize = ((struct channels_list_entry *)ptr)->datasize;
  u_int32_t bufsz = ((struct channels_list_entry *)ptr)->bufsize;
  pid_t core_pid = ((struct channels_list_entry *)ptr)->core_pid;
  struct networks_file_data nfd;

  unsigned char *rgptr;
  int pollagain = TRUE;
  u_int32_t seq = 1, rg_err_count = 0;

  struct extra_primitives extras;
  struct primitives_ptrs prim_ptrs;
  char *dataptr;

  struct p_amqp_host *amqp_host = &((struct channels_list_entry *)ptr)->amqp_host;

  memcpy(&config, cfgptr, sizeof(struct configuration));
  memcpy(&extras, &((struct channels_list_entry *)ptr)->extras, sizeof(struct extra_primitives));
  recollect_pipe_memory(ptr);
  pm_setproctitle("%s [%s]", "RabbitMQ/AMQP Plugin", config.name);

  P_set_signals();
  P_init_default_values();
  P_config_checks();
  pipebuf = (unsigned char *) pm_malloc(config.buffer_size);
  memset(pipebuf, 0, config.buffer_size);

  timeout = config.sql_refresh_time*1000;

  if (!config.sql_user) config.sql_user = rabbitmq_user;
  if (!config.sql_passwd) config.sql_passwd = rabbitmq_pwd;

  if ((config.sql_table && strchr(config.sql_table, '$')) && config.sql_multi_values) {
    Log(LOG_ERR, "ERROR ( %s/%s ): dynamic 'amqp_routing_key' is not compatible with 'amqp_multi_values'. Exiting.\n", config.name, config.type);
    exit_plugin(1);
  }

  if ((config.sql_table && strchr(config.sql_table, '$')) && config.amqp_routing_key_rr) {
    Log(LOG_ERR, "ERROR ( %s/%s ): dynamic 'amqp_routing_key' is not compatible with 'amqp_routing_key_rr'. Exiting.\n", config.name, config.type);
    exit_plugin(1);
  }

  p_amqp_init_host(&amqpp_amqp_host);
  p_amqp_set_user(&amqpp_amqp_host, config.sql_user);
  p_amqp_set_passwd(&amqpp_amqp_host, config.sql_passwd);

  /* setting function pointers */
  if (config.what_to_count & (COUNT_SUM_HOST|COUNT_SUM_NET))
    insert_func = P_sum_host_insert;
  else if (config.what_to_count & COUNT_SUM_PORT) insert_func = P_sum_port_insert;
  else if (config.what_to_count & COUNT_SUM_AS) insert_func = P_sum_as_insert;
#if defined (HAVE_L2)
  else if (config.what_to_count & COUNT_SUM_MAC) insert_func = P_sum_mac_insert;
#endif
  else insert_func = P_cache_insert;
  purge_func = amqp_cache_purge;

  memset(&nt, 0, sizeof(nt));
  memset(&nc, 0, sizeof(nc));
  memset(&pt, 0, sizeof(pt));

  load_networks(config.networks_file, &nt, &nc);
  set_net_funcs(&nt);

  if (config.ports_file) load_ports(config.ports_file, &pt);
  if (config.pkt_len_distrib_bins_str) load_pkt_len_distrib_bins();
  else {
    if (config.what_to_count_2 & COUNT_PKT_LEN_DISTRIB) {
      Log(LOG_ERR, "ERROR ( %s/%s ): 'aggregate' contains pkt_len_distrib but no 'pkt_len_distrib_bins' defined. Exiting.\n", config.name, config.type);
      exit_plugin(1);
    }
  }
  
  memset(&idata, 0, sizeof(idata));
  memset(&prim_ptrs, 0, sizeof(prim_ptrs));
  set_primptrs_funcs(&extras);

  if (config.pipe_amqp) {
    plugin_pipe_amqp_compile_check();
    pipe_fd = plugin_pipe_amqp_connect_to_consume(amqp_host, plugin_data);
    amqp_timeout = plugin_pipe_set_retry_timeout(&amqp_host->btimers, pipe_fd);
  }
  else setnonblocking(pipe_fd);

  idata.now = time(NULL);

  /* print_refresh time init: deadline */
  refresh_deadline = idata.now; 
//.........这里部分代码省略.........

static Mustache NextTag(const char *input,
                        const char *delim_start, size_t delim_start_len,
                        const char *delim_end, size_t delim_end_len)
{
    Mustache ret;
    ret.type = TAG_TYPE_NONE;

    ret.begin = strstr(input, delim_start);
    if (!ret.begin)
    {
        return ret;
    }

    ret.content = ret.begin + delim_start_len;
    char *extra_end = NULL;

    switch (ret.content[0])
    {
    case '#':
        ret.type = TAG_TYPE_SECTION;
        ret.content++;
        break;
    case '^':
        ret.type = TAG_TYPE_INVERTED;
        ret.content++;
        break;
    case '/':
        ret.type = TAG_TYPE_SECTION_END;
        ret.content++;
        break;
    case '!':
        ret.type = TAG_TYPE_COMMENT;
        ret.content++;
        break;
    case '=':
        extra_end = "=";
        ret.type = TAG_TYPE_DELIM;
        ret.content++;
        break;
    case '{':
        extra_end = "}";
    case '&':
        ret.type = TAG_TYPE_VAR_UNESCAPED;
        ret.content++;
        break;
    default:
        ret.type = TAG_TYPE_VAR;
        break;
    }

    if (extra_end)
    {
        const char *escape_end = strstr(ret.content, extra_end);
        if (!escape_end || strncmp(escape_end + 1, delim_end, delim_end_len) != 0)
        {
            Log(LOG_LEVEL_WARNING, "Broken mustache template, couldn't find end tag for quoted begin tag at '%20s'...", input);
            ret.type = TAG_TYPE_ERR;
            return ret;
        }

        ret.content_len = escape_end - ret.content;
        ret.end = escape_end + 1 + delim_end_len;
    }
    else
    {
        ret.end = strstr(ret.content, delim_end);
        if (!ret.end)
        {
            Log(LOG_LEVEL_WARNING, "Broken Mustache template, could not find end delimiter after reading start delimiter at '%20s'...", input);
            ret.type = TAG_TYPE_ERR;
            return ret;
        }

        ret.content_len = ret.end - ret.content;
        ret.end += delim_end_len;
    }

    while (*ret.content == ' ' || *ret.content == '\t')
    {
        ret.content++;
        ret.content_len--;
    }

    while (ret.content[ret.content_len - 1] == ' ' || ret.content[ret.content_len - 1] == '\t')
    {
        ret.content_len--;
    }

    return ret;
}

void amqp_cache_purge(struct chained_cache *queue[], int index)
{
  struct pkt_primitives *data = NULL;
  struct pkt_bgp_primitives *pbgp = NULL;
  struct pkt_nat_primitives *pnat = NULL;
  struct pkt_mpls_primitives *pmpls = NULL;
  char *pcust = NULL;
  struct pkt_vlen_hdr_primitives *pvlen = NULL;
  struct pkt_bgp_primitives empty_pbgp;
  struct pkt_nat_primitives empty_pnat;
  struct pkt_mpls_primitives empty_pmpls;
  char *empty_pcust = NULL;
  char src_mac[18], dst_mac[18], src_host[INET6_ADDRSTRLEN], dst_host[INET6_ADDRSTRLEN], ip_address[INET6_ADDRSTRLEN];
  char rd_str[SRVBUFLEN], misc_str[SRVBUFLEN], dyn_amqp_routing_key[SRVBUFLEN], *orig_amqp_routing_key = NULL;
  int i, j, stop, batch_idx, is_routing_key_dyn = FALSE, qn = 0, ret, saved_index = index;
  int mv_num = 0, mv_num_save = 0;
  time_t start, duration;
  pid_t writer_pid = getpid();

#ifdef WITH_JANSSON
  json_t *array = json_array();
#endif

  /* setting some defaults */
  if (!config.sql_host) config.sql_host = default_amqp_host;
  if (!config.sql_db) config.sql_db = default_amqp_exchange;
  if (!config.amqp_exchange_type) config.amqp_exchange_type = default_amqp_exchange_type;
  if (!config.amqp_vhost) config.amqp_vhost = default_amqp_vhost;

  if (!config.sql_table) config.sql_table = default_amqp_routing_key;
  else {
    if (strchr(config.sql_table, '$')) {
      is_routing_key_dyn = TRUE;
      orig_amqp_routing_key = config.sql_table;
      config.sql_table = dyn_amqp_routing_key;
    }
  }
  if (config.amqp_routing_key_rr) {
    orig_amqp_routing_key = config.sql_table;
    config.sql_table = dyn_amqp_routing_key;
  }

  p_amqp_set_exchange(&amqpp_amqp_host, config.sql_db);
  p_amqp_set_routing_key(&amqpp_amqp_host, config.sql_table);
  p_amqp_set_exchange_type(&amqpp_amqp_host, config.amqp_exchange_type);
  p_amqp_set_host(&amqpp_amqp_host, config.sql_host);
  p_amqp_set_vhost(&amqpp_amqp_host, config.amqp_vhost);
  p_amqp_set_persistent_msg(&amqpp_amqp_host, config.amqp_persistent_msg);
  p_amqp_set_frame_max(&amqpp_amqp_host, config.amqp_frame_max);
  p_amqp_set_content_type_json(&amqpp_amqp_host);

  p_amqp_init_routing_key_rr(&amqpp_amqp_host);
  p_amqp_set_routing_key_rr(&amqpp_amqp_host, config.amqp_routing_key_rr);

  empty_pcust = malloc(config.cpptrs.len);
  if (!empty_pcust) {
    Log(LOG_ERR, "ERROR ( %s/%s ): Unable to malloc() empty_pcust. Exiting.\n", config.name, config.type);
    exit_plugin(1);
  }

  memset(&empty_pbgp, 0, sizeof(struct pkt_bgp_primitives));
  memset(&empty_pnat, 0, sizeof(struct pkt_nat_primitives));
  memset(&empty_pmpls, 0, sizeof(struct pkt_mpls_primitives));
  memset(empty_pcust, 0, config.cpptrs.len);

  ret = p_amqp_connect_to_publish(&amqpp_amqp_host);
  if (ret) return;

  for (j = 0, stop = 0; (!stop) && P_preprocess_funcs[j]; j++)
    stop = P_preprocess_funcs[j](queue, &index, j);

  Log(LOG_INFO, "INFO ( %s/%s ): *** Purging cache - START (PID: %u) ***\n", config.name, config.type, writer_pid);
  start = time(NULL);

  for (j = 0; j < index; j++) {
    void *json_obj;
    char *json_str;

    if (queue[j]->valid != PRINT_CACHE_COMMITTED) continue;

    data = &queue[j]->primitives;
    if (queue[j]->pbgp) pbgp = queue[j]->pbgp;
    else pbgp = &empty_pbgp;

    if (queue[j]->pnat) pnat = queue[j]->pnat;
    else pnat = &empty_pnat;

    if (queue[j]->pmpls) pmpls = queue[j]->pmpls;
    else pmpls = &empty_pmpls;

    if (queue[j]->pcust) pcust = queue[j]->pcust;
    else pcust = empty_pcust;

    if (queue[j]->pvlen) pvlen = queue[j]->pvlen;
    else pvlen = NULL;

    if (queue[j]->valid == PRINT_CACHE_FREE) continue;

    json_obj = compose_json(config.what_to_count, config.what_to_count_2, queue[j]->flow_type,
                         &queue[j]->primitives, pbgp, pnat, pmpls, pcust, pvlen, queue[j]->bytes_counter,
//.........这里部分代码省略.........

int SelectLeaf(EvalContext *ctx, char *path, struct stat *sb, FileSelect fs)
{
    Rlist *rp;

    StringSet *leaf_attr = StringSetNew();

#ifdef __MINGW32__
    if (fs.issymlinkto != NULL)
    {
        Log(LOG_LEVEL_VERBOSE,
              "files_select.issymlinkto is ignored on Windows (symbolic links are not supported by Windows)");
    }

    if (fs.groups != NULL)
    {
        Log(LOG_LEVEL_VERBOSE,
              "files_select.search_groups is ignored on Windows (file groups are not supported by Windows)");
    }

    if (fs.bsdflags != NULL)
    {
        Log(LOG_LEVEL_VERBOSE, "files_select.search_bsdflags is ignored on Windows");
    }
#endif /* __MINGW32__ */

    if (fs.name == NULL)
    {
        StringSetAdd(leaf_attr, xstrdup("leaf_name"));
    }

    for (rp = fs.name; rp != NULL; rp = rp->next)
    {
        if (SelectNameRegexMatch(ctx, path, RlistScalarValue(rp)))
        {
            StringSetAdd(leaf_attr, xstrdup("leaf_name"));
            break;
        }
    }

    if (fs.path == NULL)
    {
        StringSetAdd(leaf_attr, xstrdup("leaf_path"));
    }

    for (rp = fs.path; rp != NULL; rp = rp->next)
    {
        if (SelectPathRegexMatch(ctx, path, RlistScalarValue(rp)))
        {
            StringSetAdd(leaf_attr, xstrdup("path_name"));
            break;
        }
    }

    if (SelectTypeMatch(sb, fs.filetypes))
    {
        StringSetAdd(leaf_attr, xstrdup("file_types"));
    }

    if ((fs.owners) && (SelectOwnerMatch(ctx, path, sb, fs.owners)))
    {
        StringSetAdd(leaf_attr, xstrdup("owner"));
    }

    if (fs.owners == NULL)
    {
        StringSetAdd(leaf_attr, xstrdup("owner"));
    }

#ifdef __MINGW32__
    StringSetAdd(leaf_attr, xstrdup("group"));

#else /* !__MINGW32__ */
    if ((fs.groups) && (SelectGroupMatch(ctx, sb, fs.groups)))
    {
        StringSetAdd(leaf_attr, xstrdup("group"));
    }

    if (fs.groups == NULL)
    {
        StringSetAdd(leaf_attr, xstrdup("group"));
    }
#endif /* !__MINGW32__ */

    if (SelectModeMatch(sb, fs.perms))
    {
        StringSetAdd(leaf_attr, xstrdup("mode"));
    }

#if defined HAVE_CHFLAGS
    if (SelectBSDMatch(sb, fs.bsdflags))
    {
        StringSetAdd(leaf_attr, xstrdup("bsdflags"));
    }
#endif

    if (SelectTimeMatch(sb->st_atime, fs.min_atime, fs.max_atime))
    {
        StringSetAdd(leaf_attr, xstrdup("atime"));
    }

//.........这里部分代码省略.........

int MQTTClient_connect(MQTTClient handle, MQTTClient_connectOptions* options)
{
	MQTTClients* m = handle;
	int rc = SOCKET_ERROR;
	START_TIME_TYPE start;
	long millisecsTimeout = 30000L;

	FUNC_ENTRY;
	Thread_lock_mutex(mqttclient_mutex);

	if (options == NULL)
	{
		rc = MQTTCLIENT_NULL_PARAMETER;
		goto exit;
	}

	if (strncmp(options->struct_id, "MQTC", 4) != 0 || (options->struct_version != 0 && options->struct_version != 1))
	{
		rc = MQTTCLIENT_BAD_STRUCTURE;
		goto exit;
	}

	if (options->will) /* check validity of will options structure */
	{
		if (strncmp(options->will->struct_id, "MQTW", 4) != 0 || options->will->struct_version != 0)
		{
			rc = MQTTCLIENT_BAD_STRUCTURE;
			goto exit;
		}
	}

#if defined(OPENSSL)
	if (options->struct_version != 0 && options->ssl) /* check validity of SSL options structure */
	{
		if (strncmp(options->ssl->struct_id, "MQTS", 4) != 0 || options->ssl->struct_version != 0)
		{
			rc = MQTTCLIENT_BAD_STRUCTURE;
			goto exit;
		}
	}
#endif

	if ((options->username && !UTF8_validateString(options->username)) ||
		(options->password && !UTF8_validateString(options->password)))
	{
		rc = MQTTCLIENT_BAD_UTF8_STRING;
		goto exit;
	}

	millisecsTimeout = options->connectTimeout * 1000;
	start = MQTTClient_start_clock();
	if (m->ma && !running)
	{
		Thread_start(MQTTClient_run, handle);
		if (MQTTClient_elapsed(start) >= millisecsTimeout)
		{
			rc = SOCKET_ERROR;
			goto exit;
		}
		MQTTClient_sleep(100L);
	}

	m->c->keepAliveInterval = options->keepAliveInterval;
	m->c->cleansession = options->cleansession;
	m->c->maxInflightMessages = (options->reliable) ? 1 : 10;

	if (options->will && options->will->struct_version == 0)
	{
	    m->c->will = malloc(sizeof(willMessages));
		m->c->will->msg = options->will->message;
		m->c->will->qos = options->will->qos;
		m->c->will->retained = options->will->retained;
		m->c->will->topic = options->will->topicName;
	}

#if defined(OPENSSL)
	if (options->struct_version != 0 && options->ssl)
		m->c->sslopts = options->ssl;
#endif

	m->c->username = options->username;
	m->c->password = options->password;
	m->c->retryInterval = options->retryInterval;

	Log(TRACE_MIN, -1, "Connecting to serverURI %s", m->serverURI);
#if defined(OPENSSL)
	rc = MQTTProtocol_connect(m->serverURI, m->c, m->ssl);
#else
	rc = MQTTProtocol_connect(m->serverURI, m->c);
#endif
	if (rc == SOCKET_ERROR)
		goto exit;

	if (m->c->connect_state == 0)
	{
		rc = SOCKET_ERROR;
		goto exit;
	}

	if (m->c->connect_state == 1) /* TCP connect started - wait for completion */
//.........这里部分代码省略.........