Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.logger.debug(
"[%s:%s] Connected!", self.fn or self.host, self.port
)
return
except OSError as err:
self.connecting = True
if self.stop.is_set():
self.logger.error(
"[%s:%s] Failed to connect: %s. "
"aborting due to stop signal.",
self.fn or self.host,
self.port,
err,
)
raise ChromecastConnectionError("Failed to connect")
self._report_connection_status(
ConnectionStatus(
CONNECTION_STATUS_FAILED,
NetworkAddress(self.host, self.port),
)
)
if service is not None:
retry_log_fun(
"[%s:%s] Failed to connect to service %s"
", retrying in %.1fs",
self.fn or self.host,
self.port,
service,
retry["delay"],
)
"[%s:%s] Not connected, sleeping for %.1fs. Services: %s",
self.fn or self.host,
self.port,
self.retry_wait,
self.services,
)
time.sleep(self.retry_wait)
if tries:
tries -= 1
self.stop.set()
self.logger.error(
"[%s:%s] Failed to connect. No retries.", self.fn or self.host, self.port
)
raise ChromecastConnectionError("Failed to connect")
def _send_protocol(self, data):
""" Default handler for sending messages as subprotocol. """
if _DEBUG:
self.logger.info("Sending {}".format(data))
if not self.client:
raise error.ChromecastConnectionError(
"Not connected to Chromecast")
try:
self.client.send(json.dumps([self.protocol, data]).encode("utf8"))
except socket.error:
# if an error occured sending data over the socket
raise error.ChromecastConnectionError(
"Error communicating with Chromecast")
def _send_protocol(self, data):
""" Default handler for sending messages as subprotocol. """
if _DEBUG:
self.logger.info("Sending {}".format(data))
if not self.client:
raise error.ChromecastConnectionError(
"Not connected to Chromecast")
try:
self.client.send(json.dumps([self.protocol, data]).encode("utf8"))
except socket.error:
# if an error occured sending data over the socket
raise error.ChromecastConnectionError(
"Error communicating with Chromecast")