Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .github/workflows/go.yml
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,9 @@ jobs:
- name: Build
run: go build -v ./...

- name: Build JS/WASM
run: GOOS=js GOARCH=wasm go build ./...

- name: Test
run: go test -v ./...

Expand Down
5 changes: 5 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,11 @@ All notable HyperMeow changes are documented here. HyperMeow uses commit pseudo-

## [Unreleased]

### Added

- Opt-in concurrent WebSocket connections through `SocketConfig.RaceURLs`, with asynchronous loser cleanup and support for native and JS/WASM builds. See [connection options](README.md#connection-options).
- `Client.DisablePostConnectPassiveIQ` to skip the post-connect passive/active IQ when the login payload already declares an active connection.

### Fixed

- Keep connection handler completion pending until an active handler exits, and process retired stream errors before automatic reconnect decisions. Manual disconnect retains its five-second wait bound.
Expand Down
39 changes: 39 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,45 @@ Custom `store.ChatSettingsStore` implementations must add `PutWASARootSecretID`
and `GetWASARootSecretID` for bot root-secret state. `sqlstore` adds this field
through schema version 19; its existing migrations remain unchanged.

## Connection options

Connections use one WebSocket endpoint by default. To try both WhatsApp chat
endpoints concurrently, configure the client before connecting:

```go
import (
whatsmeow "github.com/polymorfa/hypermeow"
"github.com/polymorfa/hypermeow/socket"
)

// client is an existing *whatsmeow.Client.
client.SocketConfig = &whatsmeow.SocketConfig{
RaceURLs: append([]string(nil), socket.RaceURLs...),
}
```

`SocketConfig.RaceURLs` activates racing when it contains at least two URLs.
The first successful upgrade wins. Pending upgrades are canceled, and opened
losers close asynchronously with status 1000 and reason `loser socket`.
The winner proceeds without waiting for those close handshakes. If every
endpoint fails, the returned error joins the failures in configured URL order.
Initial background reconnect classification uses the first matching HTTP status
in configured URL order, ahead of network errors from other endpoints. When no
HTTP status is present, network and dial failures remain retryable. The existing
single-endpoint retry rules remain unchanged.

With zero or one race URL, the client uses `SocketConfig.URL`, or the default
endpoint if `URL` is empty. With racing enabled, the list supplies the endpoints
and `URL` is ignored. `SocketConfig.Origin` and `Client.WebSocketHeaders` apply to
all racers. Repeated header values and empty entries that suppress defaults are
preserved. Browsers control request headers on JS/WASM builds, as they do for
single-endpoint connections.

`Client.DisablePostConnectPassiveIQ` defaults to `false`. Set it to `true` only
when your login payload already declares `passive: false`; the client then skips
the post-connect passive/active IQ. This option does not change the login payload.
Configure both options before connecting rather than while a connection is active.

## Discord

Join the [WhatsApp Web ecosystem Discord](https://whiskey.so/discord) and visit
Expand Down
37 changes: 31 additions & 6 deletions client.go
Original file line number Diff line number Diff line change
Expand Up @@ -282,6 +282,11 @@ type Client struct {
// The user agent to use (for non-Messenger connections).
UserAgent string
WebSocketHeaders http.Header
// DisablePostConnectPassiveIQ stops the library sending the
// `passive`/`active` IQ after a successful connect. Set it when the
// client payload already carries `passive: false`, which is what
// WhatsApp Web sends; the real client has no such IQ.
DisablePostConnectPassiveIQ bool
}

type groupMetaCache struct {
Expand All @@ -301,6 +306,15 @@ type SocketConfig struct {
URL string
Origin string
NoiseCertificateAuthority *[32]byte
// RaceURLs, when it has more than one entry, makes Connect open every
// URL concurrently and keep the one that connects first, closing the
// others with code 1000 and the reason "loser socket".
//
// Use socket.RaceURLs for the endpoints WhatsApp Web races
// (WAWebOpenSocket.js:10, 44-52). Loser cleanup runs asynchronously.
//
// Zero or one entry keeps the single-socket behavior and uses URL above.
RaceURLs []string
}

const handlerQueueSize = 256
Expand Down Expand Up @@ -563,10 +577,8 @@ func (cli *Client) Connect() error {
}

func isRetryableConnectError(err error) bool {
if exhttp.IsNetworkError(err) {
return true
}

// HTTP upgrade rejections take precedence over network failures from
// other racers. Joined errors preserve the configured endpoint order.
var statusErr socket.ErrWithStatusCode
if errors.As(err, &statusErr) {
switch statusErr.StatusCode {
Expand All @@ -577,6 +589,9 @@ func isRetryableConnectError(err error) bool {
}
}

if exhttp.IsNetworkError(err) {
return true
}
return errors.Is(err, socket.ErrDialFailed)
}

Expand Down Expand Up @@ -641,18 +656,28 @@ func (cli *Client) unlockedConnect(ctx context.Context) error {
}
var queue chan *waBinary.Node
maps.Copy(fs.HTTPHeaders, cli.WebSocketHeaders)
var raceURLs []string
if cli.SocketConfig != nil {
if cli.SocketConfig.URL != "" {
fs.URL = cli.SocketConfig.URL
}
if cli.SocketConfig.Origin != "" {
fs.HTTPHeaders.Set("Origin", cli.SocketConfig.Origin)
}
raceURLs = cli.SocketConfig.RaceURLs
}
if err := fs.Connect(ctx); err != nil {
if len(raceURLs) > 1 {
raced, err := socket.ConnectRace(ctx, cli.Log.Sub("Socket"), client, raceURLs, fs.HTTPHeaders)
Comment thread
purpshell marked this conversation as resolved.
if err != nil {
return err
}
fs = raced
} else if err := fs.Connect(ctx); err != nil {
fs.Close(0)
return err
} else if queue, err = cli.doHandshake(fs, *keys.NewKeyPair()); err != nil {
}
var err error
if queue, err = cli.doHandshake(fs, *keys.NewKeyPair()); err != nil {
fs.Close(0)
return fmt.Errorf("noise handshake failed: %w", err)
}
Expand Down
13 changes: 10 additions & 3 deletions connectionevents.go
Original file line number Diff line number Diff line change
Expand Up @@ -197,9 +197,16 @@ func (cli *Client) handleConnectSuccess(ctx context.Context, node *waBinary.Node
cli.Log.Debugf("Prekey count after upload: %d", sc)
}
}
err := cli.SetPassive(ctx, false)
if err != nil {
cli.Log.Warnf("Failed to send post-connect passive IQ: %v", err)
// WhatsApp Web sends `passive: false` in the login payload itself
// (WAWebGetClientPayloadForLogin.js:14-19) and never sends this IQ.
// A client that logs in passive and then immediately asks to become
// active does something no real page does, so a caller that already
// carries the right value in its payload turns this off.
if !cli.DisablePostConnectPassiveIQ {
err := cli.SetPassive(ctx, false)
if err != nil {
cli.Log.Warnf("Failed to send post-connect passive IQ: %v", err)
}
}
cli.dispatchEvent(&events.Connected{})
cli.closeSocketWaitChan()
Expand Down
57 changes: 47 additions & 10 deletions socket/framesocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,17 @@ func (fs *FrameSocket) SetOnDisconnect(handler func(ctx context.Context, remote
}

func (fs *FrameSocket) Close(code websocket.StatusCode) {
fs.CloseWithReason(code, "")
}

// CloseWithReason closes the socket with an explicit close reason.
//
// WhatsApp Web races two websockets on every connect and closes the one that
// loses with code 1000 and the reason "loser socket"
// (WAWebOpenSocket.js:44-52). A close with an empty reason where the client
// sends one is observable, so the reason is part of the wire behaviour and not
// a log detail.
func (fs *FrameSocket) CloseWithReason(code websocket.StatusCode, reason string) {
fs.lock.Lock()
defer fs.lock.Unlock()

Expand All @@ -84,7 +95,7 @@ func (fs *FrameSocket) Close(code websocket.StatusCode) {

fs.closed.Store(true)
if code > 0 {
err := conn.Close(code, "")
err := conn.Close(code, reason)
if err != nil {
fs.log.Warnf("Error sending close to websocket: %v", err)
}
Expand All @@ -102,16 +113,24 @@ func (fs *FrameSocket) Close(code websocket.StatusCode) {
}

func (fs *FrameSocket) Connect(ctx context.Context) error {
return fs.dial(ctx, ctx, true)
}

// dial separates the lifetime of the opened socket from the context used
// for the HTTP upgrade. Most callers use the same context for both through
// Connect. Socket racing uses a short-lived dial context so aborting another
// in-flight upgrade cannot cancel a connection that has already opened.
func (fs *FrameSocket) dial(parentCtx, dialCtx context.Context, startPump bool) error {
fs.lock.Lock()
defer fs.lock.Unlock()
if fs.conn.Load() != nil {
return ErrSocketAlreadyOpen
}
fs.parentCtx = ctx
fs.cancelCtx, fs.cancel = context.WithCancel(ctx)
fs.parentCtx = parentCtx
fs.cancelCtx, fs.cancel = context.WithCancel(parentCtx)

fs.log.Debugf("Dialing %s", fs.URL)
conn, resp, err := websocket.Dial(ctx, fs.URL, fs.makeDialOptions())
conn, resp, err := websocket.Dial(dialCtx, fs.URL, fs.makeDialOptions())
if err != nil {
if resp != nil {
err = ErrWithStatusCode{err, resp.StatusCode}
Expand All @@ -122,11 +141,20 @@ func (fs *FrameSocket) Connect(ctx context.Context) error {
conn.SetReadLimit(FrameMaxSize)

fs.conn.Store(conn)

go fs.readPump(conn, ctx)
if startPump {
go fs.readPump(conn, fs.cancelCtx)
}
return nil
}

func (fs *FrameSocket) startReadPump() {
fs.lock.Lock()
defer fs.lock.Unlock()
if conn := fs.conn.Load(); conn != nil {
go fs.readPump(conn, fs.cancelCtx)
}
}

func (fs *FrameSocket) Context() context.Context {
return fs.cancelCtx
}
Expand Down Expand Up @@ -163,13 +191,18 @@ func (fs *FrameSocket) SendFrame(data []byte) error {
return conn.Write(fs.cancelCtx, websocket.MessageBinary, wholeFrame)
}

func (fs *FrameSocket) frameComplete() {
func (fs *FrameSocket) frameComplete() bool {
data := fs.incoming
fs.incoming = nil
fs.partialHeader = nil
fs.incomingLength = 0
fs.receivedLength = 0
fs.Frames <- data
select {
case fs.Frames <- data:
return true
case <-fs.cancelCtx.Done():
return false
}
}

func (fs *FrameSocket) processData(msg []byte) {
Expand All @@ -188,7 +221,9 @@ func (fs *FrameSocket) processData(msg []byte) {
if len(msg) >= length {
fs.incoming = msg[:length]
msg = msg[length:]
fs.frameComplete()
if !fs.frameComplete() {
return
}
} else {
fs.incoming = make([]byte, length)
copy(fs.incoming, msg)
Expand All @@ -203,7 +238,9 @@ func (fs *FrameSocket) processData(msg []byte) {
if fs.receivedLength+len(msg) >= fs.incomingLength {
copy(fs.incoming[fs.receivedLength:], msg[:fs.incomingLength-fs.receivedLength])
msg = msg[fs.incomingLength-fs.receivedLength:]
fs.frameComplete()
if !fs.frameComplete() {
return
}
} else {
copy(fs.incoming[fs.receivedLength:], msg)
fs.receivedLength += len(msg)
Expand Down
Loading
Loading