Websocket.cpp

#include "Websocket.h"

#if defined(ARDUINO)
void WebsocketClient::enableHeartbeat(uint32_t interval, uint32_t timeout, uint8_t maxMissed) {
  WebSocketsClient::enableHeartbeat(interval, timeout, maxMissed);
}

void WebsocketClient::disableHeartbeat() {
  WebSocketsClient::disableHeartbeat();
}

void WebsocketClient::setReconnectInterval(uint32_t interval) {
  reconnectInterval = interval;
  reconnectDelay = interval;
  reconnectLastAttempt = millis()-interval;
  // Reconnection pacing is managed here so failed attempts can back off.
  WebSocketsClient::setReconnectInterval(0);
}

void WebsocketClient::resetStreaming() {
  isStreaming = false;
}

void WebsocketClient::poll() {
  if (_client.status == WSC_NOT_CONNECTED) {
    uint32_t now = millis();
    if (now-reconnectLastAttempt < reconnectDelay) {
      return;
    }

    reconnectLastAttempt = now;
    WebSocketsClient::loop();
    if (_client.status == WSC_NOT_CONNECTED) {
      if (reconnectDelay < 15000) {
        reconnectDelay = 15000;
      } else if (reconnectDelay < 60000) {
        reconnectDelay = 60000;
      } else if (reconnectDelay < 300000) {
        reconnectDelay = 300000;
      }
    }
    return;
  }

  WebSocketsClient::loop();
  if (_client.status == WSC_CONNECTED) {
    reconnectDelay = reconnectInterval;
  } else if (_client.status == WSC_NOT_CONNECTED) {
    reconnectDelay = reconnectInterval;
    reconnectLastAttempt = millis();
  }
}

void WebsocketClient::onEvent(WebSocketEventCallback callback) {
  WS_DEBUG("Setting event callback\n");
  this->eventCallback = callback;
}

void WebsocketClient::connect(WSInterfaceString host, int port, WSInterfaceString path) {
  WS_DEBUG("Connecting to ws://%s:%d%s\n", host.c_str(), port, path.c_str());
  WebSocketsClient::begin(host, port, path);
}

void WebsocketClient::connectSecure(WSInterfaceString host, int port, WSInterfaceString path) {
  WebSocketsClient::beginSSL(host.c_str(), port, path.c_str());
}

bool WebsocketClient::stream() {
  if (isStreaming) {
    WS_DEBUG("Already streaming\n");
    return false;
  }

  if (clientIsConnected(&_client)) {
    isStreaming = sendFrame(&_client, WSop_text, (uint8_t *) "", 0, false, false);
  } else {
    isStreaming = false;
  }

  return isStreaming;
}

bool WebsocketClient::send(uint8_t *payload, size_t length, bool headerToPayload) {
  WS_DEBUG("Sending message of length %d\n", length);

  if (length == 0) {
    length = strlen((const char *) payload);
  }

  if (clientIsConnected(&_client)) {
    if (isStreaming) {
      return sendFrame(&_client, WSop_continuation, payload, length, false, headerToPayload);
    } else {
      return sendFrame(&_client, WSop_text, payload, length, true, headerToPayload);
    }
  }

  return false;
}

bool WebsocketClient::send(const char *payload, size_t length, bool headerToPayload) {
  return send((uint8_t *) payload, length, headerToPayload);
}

bool WebsocketClient::end() {
  if (!isStreaming) {
    return true;
  }

  WS_DEBUG("Ending stream\n");

  bool res = sendFrame(&_client, WSop_continuation, (uint8_t *) "", 0, true, false);
  isStreaming = !res;
  return res;
}

#else

void WebsocketClient::enableHeartbeat(uint32_t interval, uint32_t timeout, uint8_t maxMissed) {
    heartbeatEnabled = true;
    heartbeatInterval = interval;
    heartbeatTimeout = timeout;
    heartbeatMaxMissed = maxMissed;
}

void WebsocketClient::disableHeartbeat() {
    heartbeatEnabled = false;
}

void WebsocketClient::setReconnectInterval(uint32_t interval) {
    reconnectInterval = interval;
}

void WebsocketClient::resetStreaming() {
    isStreaming = false;
}

// Arduino-compatible millis() for non-Arduino targets: ms since first call,
// returned as uint32_t so it matches Arduino's 32-bit wrap semantics
// (every ~49 days). OTF carries its own epoch state so the library stays
// standalone.
uint32_t millis() {
	static uint64_t epoch_ms = 0;
	struct timeval tv;
	gettimeofday(&tv, NULL);
	uint64_t now = (uint64_t)tv.tv_sec * (uint64_t)1000 + (uint64_t)(tv.tv_usec / 1000);
	if (epoch_ms == 0) epoch_ms = now;
	return (uint32_t)(now - epoch_ms);
}


void WebsocketClient::poll() {
  websockets::WebsocketsClient::poll();
  if (heartbeatEnabled && available()) {
    if (!heartbeatInProgress && (millis() - heartbeatLastSent > heartbeatInterval)) {
      if (heartbeatMissed >= heartbeatMaxMissed) {
        // Too many missed heartbeats, close the connection
        WS_DEBUG("Too many missed heartbeats, closing connection\n");
        reconnectLastAttempt = 0;
        heartbeatMissed = 0;
        websockets::WebsocketsClient::close();
        return;
      }

      WS_DEBUG("Sending ping\n");
      ping();
      heartbeatLastSent = millis();
      heartbeatInProgress = true;
    }

    if (heartbeatInProgress && (millis() - heartbeatLastSent > heartbeatTimeout)) {
      // Heartbeat timeout
      WS_DEBUG("Heartbeat timeout\n");
      heartbeatMissed++;
      heartbeatInProgress = false;
      return;
    }
  }

  if (shouldReconnect && !available()) {
    if (millis() - reconnectLastAttempt > reconnectInterval) {
      WS_DEBUG("Reconnecting...\n");
      // Attempt to reconnect
        websockets::WebsocketsClient::connect(host, port, path);

      WS_DEBUG("Reconnect attempt complete\n");
      WS_DEBUG("Connection status: %d\n", websockets::WebsocketsClient::available());
      reconnectLastAttempt = millis();
    }
  }
}

void WebsocketClient::onEvent(WebSocketEventCallback callback) {
  WS_DEBUG("Setting event callback\n");
  this->eventCallback = callback;
}

void WebsocketClient::connect(WSInterfaceString host, int port, WSInterfaceString path) {
  WS_DEBUG("Connecting to ws://%s:%d%s\n", host.c_str(), port, path.c_str());
  this->host = host;
  this->port = port;
  this->path = path;
  shouldReconnect = true;
  heartbeatMissed = 0;
  heartbeatInProgress = false;
//   isSecure = false;
  websockets::WebsocketsClient::connect(host, port, path);
}

// void WebsocketClient::connectSecure(WSInterfaceString host, int port, WSInterfaceString path) {
//   WS_DEBUG("Connecting to wss://%s:%d%s\n", host.c_str(), port, path.c_str());
//   this->host = host;
//   this->port = port;
//   this->path = path;
//   shouldReconnect = true;
//   heartbeatMissed = 0;
//   heartbeatInProgress = false;
//   isSecure = true;
//   websockets::WebsocketsClient::connect(host, port, path);
// }

bool WebsocketClient::stream() {
  return websockets::WebsocketsClient::stream();
}

bool WebsocketClient::send(uint8_t *payload, size_t length, bool headerToPayload) {
    return send((const char*) payload, length, headerToPayload);
}

bool WebsocketClient::send(const char *payload, size_t length, bool headerToPayload) {
    WS_DEBUG("Sending message of length %d\n", length);
    if (length == 0) {
        length = strlen(payload);
    }

  return websockets::WebsocketsClient::send((const char*) payload, length);
}

bool WebsocketClient::end() {
  return websockets::WebsocketsClient::end();
}

#endif