CSiteCheckerTest::check_pull_stream

function check_pull_stream()
{
	if (CModule::IncludeModule('pull'))
	{
		$text = md5(mt_rand(100000,999999));
		$channelId = md5($text);
		$id = CPullChannel::SignChannel($channelId);
		if (CPullOptions::GetQueueServerStatus())
		{
			$isServerShared = CPullOptions::IsServerShared();
			if($isServerShared && !BitrixPullSharedServerConfig::isRegistered())
			{
				$this->arTestVars['push_stream_fail'] = true;
				return $this->Result(false, GetMessage("MAIN_SC_PULL_NOT_REGISTERED"));
			}
			else if (CPullOptions::GetQueueServerVersion() < 4)
			{
				$this->arTestVars['push_stream_warn'] = true;
				return $this->Result(null, GetMessage("MAIN_SC_PULL_UNSUPPORTED_VERSION"));
			}
			else
			{
				if (!$ar = parse_url(str_replace('#DOMAIN#', $this->host, BitrixPullConfig::getPublishUrl($id))))
					return $this->Result(false, GetMessage("MAIN_SC_PATH_PUB"));

				$pub_domain = $ar['host'];
				$pub_host = ($ar['scheme'] == 'https' ? 'ssl://' : '').$pub_domain;
				$pub = $ar['path'].'?'.$ar['query'];
				$pub_port = $ar['port'];
				if (!$pub_port)
					$pub_port = $ar['scheme'] == 'https' ? 443 : 80;

				if($isServerShared)
				{
					$listenUrl = BitrixPullSharedServerConfig::getLongPollingUrl();
					$listenUrl .= "?CHANNEL_ID=".$id . "&clientId=".BitrixPullSharedServerClient::getPublicLicenseCode();
				}
				else
				{
					$listenUrl = $this->ssl ? CPullOptions::GetListenSecureUrl($id) : CPullOptions::GetListenUrl($id);
				}

				if (!$ar = parse_url(str_replace('#DOMAIN#', $this->host, $listenUrl)))
					return $this->Result(false, GetMessage("MAIN_SC_PATH_SUB"));

				$sub_domain = $ar['host'];
				$sub_host = ($ar['scheme'] == 'https' ? 'ssl://' : '').$sub_domain;
				$sub = $ar['path'].'?'.$ar['query'];
				$sub_port = $ar['port'];
				if (!$sub_port)
					$sub_port = $ar['scheme'] == 'https' ? 443 : 80;
			}
		}
		else
		{
			$this->arTestVars['push_stream_fail'] = true;
			return $this->Result(false, GetMessage("MAIN_SC_STREAM_DISABLED_2"));
		}
	}
	else
	{
		$this->arTestVars['push_stream_fail'] = true;
		return $this->Result(false, GetMessage("MAIN_NO_PULL"));
	}

	$ver = CPullOptions::GetQueueServerVersion();
	$bNodeJS = $ver > 2;
	echo 'Server version: '.$ver.($bNodeJS ? ' (Bitrix Push server)' : ' (nginx-push-stream-module)')."n";

	$strRequest0 = 'POST '.$pub.' HTTP/1.0'."rn".
		'Host: '.$pub_domain."rn".
		'Content-Length: '.strlen($text)."rn".
		"rn".
		$text."rn";
	$strRequest1 = 'GET '.$sub.' HTTP/1.0'."rn".
		'If-Modified-Since: '.date('r', time() - 86400)."rn".
		'Host: '.$sub_domain."rnrn";
	$strRequest2 = 'DELETE '.$pub.' HTTP/1.0'."rn".
		'Host: '.$sub_domain."rnrn";

	if (!$bNodeJS)
	{
		// POST - to create a channel
		if (!$res0 = $this->ConnectToHost($pub_host, $pub_port))
		{
			$this->arTestVars['push_stream_fail'] = true;
			return $this->Result(false, GetMessage("MAIN_SC_NO_PUSH_STREAM_CONNECTION"));
		}
		fwrite($res0, $strRequest0);
		fclose($res0);
	}

	// GET - connection
	if (!$res1 = $this->ConnectToHost($sub_host, $sub_port))
	{
		$this->arTestVars['push_stream_fail'] = true;
		return $this->Result(false, GetMessage("MAIN_SC_NO_SUB_CONNECTION_2"));
	}
	fwrite($res1, $strRequest1);
	sleep(1); // we need some time to create channel

	// POST - message
	$postResult = CPullChannel::Send($channelId, $text, ['dont_wait_answer' => false]);
	if(!$postResult)
	{
		$this->arTestVars['push_stream_fail'] = true;
		return $this->Result(false, GetMessage("MAIN_SC_NO_PUSH_STREAM_CONNECTION"));
	}

	// GET - message
	$strRes1 = fread($res1, 4096);

	$retVal = true;
	if (false === strpos($strRes1, $text))
	{
//			PrintHTTP($strRequest0, $strHeaders0, $strRes0);
		PrintHTTP($strRequest1, $strHeaders1, $strRes1);
		$this->arTestVars['push_stream_fail'] = true;
		$retVal = $this->Result(false, GetMessage("MAIN_SC_PUSH_INCORRECT", ['#MODULE#' => $bNodeJS ? 'Bitrix Push server' : 'nginx-push-stream-module']));
	}

	// DELETE
	if (!$res2 = $this->ConnectToHost($pub_host, $pub_port))
	{
		$this->arTestVars['push_stream_fail'] = true;
		return $this->Result(false, GetMessage("MAIN_SC_NO_PUSH_STREAM_CONNECTION_2"));
	}
	fwrite($res2, $strRequest2);
	fclose($res2);

	if ($retVal == true && COption::GetOptionString('main', 'session_expand', 'Y') <> 'N' && (!defined("BX_SKIP_SESSION_EXPAND") || BX_SKIP_SESSION_EXPAND === false))
		return $this->Result(null, GetMessage('MAIN_SC_WARN_EXPAND_SESSION'));

	return $retVal;
}