Error [ERR_STREAM_PREMATURE_CLOSE]: Premature close

The goal is to get live stream notification, when the profile which I follow starts a live stream. And when some one send a direct message to me, I can get a notification. I am a rookie, so I may have made some low-level mistakes. sorry for that.

I login the account A.

Problem1: I use the example/push.example.ts & realtime.example.ts, but it always shows the ERROR with ‘Premature Close’, and when I use account B to send message to account A, there is no message show in the console.

Problem2: I use readState(ig) and saveState(ig), but the ‘new activity’ in instagram app shows multi login activities. I thought there should be one login activity because after the first login from a never-used device, the second login uses the cookie from the state.json.

 ig:mqtt:realtime Subscribing with Skywalker to ig/u/v1/******, ig/live_notification_subscribe/******** +0ms
  ig:mqtt:realtime Iris Sub to: seqId: ****, snapshot: ****** +0ms
  ig:mqtt:mqttot edge-mqtt.facebook.com: Error: Premature close
  ig:mqtt:mqttot        Stack: Error [ERR_STREAM_PREMATURE_CLOSE]: Premature close
  ig:mqtt:mqttot     at new NodeError (node:internal/errors:329:5)
  ig:mqtt:mqttot     at MqttTransformer.onclose (node:internal/streams/end-of-stream:138:38)
  ig:mqtt:mqttot     at MqttTransformer.emit (node:events:381:22)
  ig:mqtt:mqttot     at MqttTransformer.emit (node:domain:470:12)
  ig:mqtt:mqttot     at emitCloseNT (node:internal/streams/destroy:169:10)
  ig:mqtt:mqttot     at processTicksAndRejections (node:internal/process/task_queues:80:21) +0ms
  ig:mqtt:mqttot edge-mqtt.facebook.com: Warning: _a.check is not a function
  ig:mqtt:mqttot        Stack: TypeError: _a.check is not a function
  ig:mqtt:mqttot     at MQTToTClient.setDisconnected (/project/path/node_modules/mqtts/src/mqtt.client.ts:475:55)
  ig:mqtt:mqttot     at /project/path/node_modules/mqtts/src/mqtt.client.ts:175:33
  ig:mqtt:mqttot     at node:internal/util:408:5
  ig:mqtt:mqttot     at finish (node:internal/streams/pipeline:163:7)
  ig:mqtt:mqttot     at node:internal/util:408:5
  ig:mqtt:mqttot     at node:internal/streams/pipeline:74:5
  ig:mqtt:mqttot     at finish (node:internal/streams/pipeline:159:23)
  ig:mqtt:mqttot     at node:internal/util:408:5
  ig:mqtt:mqttot     at node:internal/streams/pipeline:74:5
  ig:mqtt:mqttot     at finish (node:internal/streams/pipeline:159:23)
  ig:mqtt:mqttot     at node:internal/util:408:5
  ig:mqtt:mqttot     at MqttTransformer.<anonymous> (node:internal/streams/pipeline:66:7)
  ig:mqtt:mqttot     at MqttTransformer.<anonymous> (node:internal/util:408:5)
  ig:mqtt:mqttot     at MqttTransformer.onclose (node:internal/streams/end-of-stream:138:25)
  ig:mqtt:mqttot     at MqttTransformer.emit (node:events:381:22)
  ig:mqtt:mqttot     at MqttTransformer.emit (node:domain:470:12)
  ig:mqtt:mqttot     at emitCloseNT (node:internal/streams/destroy:169:10)
  ig:mqtt:mqttot     at processTicksAndRejections (node:internal/process/task_queues:80:21) +0ms

import 'dotenv/config';
import { IgApiClient } from 'instagram-private-api';
import { IgApiClientExt, withRealtime } from 'instagram_mqtt';
import { GraphQLSubscriptions } from 'instagram_mqtt';
import { SkywalkerSubscriptions } from 'instagram_mqtt';
import { promisify } from 'util';
import { writeFile, readFile, exists } from 'fs';

const writeFileAsync = promisify(writeFile);
const readFileAsync = promisify(readFile);
const existsAsync = promisify(exists);

async function saveState(ig: IgApiClientExt) {
    return writeFileAsync('state.json', await ig.exportState(), {encoding: 'utf8'});
}

async function readState(ig: IgApiClientExt) {
    if (!await existsAsync('state.json'))
        return;
    await ig.importState(await readFileAsync('state.json', {encoding: 'utf8'}));
}

async function loginToInstagram(ig: IgApiClientExt) {
    ig.request.end$.subscribe(() => saveState(ig));
    await ig.account.login(process.env.IG_USERNAME, process.env.IG_PASSWORD);
}


(async () => {
    const ig = withRealtime(new IgApiClient());
    ig.state.generateDevice(process.env.IG_USERNAME);
    ig.state.proxyUrl = process.env.IG_PROXY;

    // this will set the auth and the cookies for instagram
    await readState(ig);

    // this logs the client in
    // await ig.simulate.preLoginFlow();
    await loginToInstagram(ig);

    await ig.realtime.connect({
        irisData: await ig.feed.directInbox().request(),
        socksOptions: {
            type: 5,
            port: 6153,
            host: '127.0.0.1'
        }
    });

    function logEvent(name: string) {
        return (data: any) => console.log(name, data);
    }

    ig.realtime.on('message', (message) => console.log(message));
    ig.realtime.on('receive', (topic, messages) => {
        console.log('receive', topic, messages);
    });
    ig.realtime.on('error', console.error);
    ig.realtime.on('direct', logEvent('direct'));
})();