socket-server

app.js

const sp = require('socket.io-pull-stream');
const pull = require('pull-stream');
const io = require('socket.io-client');

const socket = io('http://localhost:8000');
// add functions
sp(socket);

socket.on('connect', () => {
  // console.log('id:', socket.id);
});

const dealEachObject = pull.asyncMap(async (readings, cb) => {
  console.log('readings:', readings);
  cb(null, readings);
});

const afterAll = (error) => {
  if (error) {
    throw error;
  }
};


socket.on('file', (id) => {
  console.log('id:', id);
  pull(
    socket.createSource(id, { codec: 'plain' }), // to recieve just create a source with the id (IMPORTANT: DO THAT SYNC)
    dealEachObject,
    pull.drain(null, afterAll)
    // pull.collect((err, data) => {
    //   if (err) throw err;
    //   console.log('data:', data)
    // })
  );
});

server.js

const sp = require('socket.io-pull-stream');
const fs = require('fs');
const toPull = require('stream-to-pull-stream');
const pull = require('pull-stream');
const csv = require('csv-parser')
const http = require('http');
const socketio = require('socket.io');

// test with browser localhost:8000
const server = http.createServer((req, res) => {
  res.end('I am connected');
});

// io server
const io = socketio(server, {
  cors: {
    origin: '*',
    methods: ['GET', 'POST'],
  },
});



// pull-stream
io.on('connection', (client) => {
  // add functions to socketio client instance
  sp(client);
  const source = toPull.source(fs.createReadStream('./test.csv').pipe(csv()));

  const sink = client.createSink('123', { codec: 'plain' });
  // console.log('sink:', sink)

  client.emit('file', sink.id) // to send the stream just emit the id (IMPORTANT: EMIT THE ID FIRST, LATER CREATE THE SOURCE IN SYNC)
  
  pull(
    source,
    sink
  )
});

server.listen(8000);