Real-Time Location Tracking With Change Streams and Socket.io

1
// dbClient.ts
2
import { MongoClient } from 'mongodb';
3
const uri = process.env.MONGODB_CONNECTION_STRING;
4
let db: MongoClient;
5
const dbConnect = async (): Promise<MongoClient> => {
6
    try {
7
        if (db) {
8
            return db;
9
        }
10
        console.log('Connecting to MongoDB...');
11
        const client = new MongoClient(uri);
12
        await client.connect();
13
        db = client;
14
        console.log('Connected to db');
15
        return db;
16
    } catch (error) {
17
        console.error('Error connecting to MongoDB', error);
18
        throw error;
19
    }
20
};
21
export default dbConnect;

1
// server.ts
2
import dbClient from './dbClient';
3
server.listen(5000, async () => {
4
    try {
5
        await dbClient();
6
    } catch (error) {
7
        console.error(error);
8
    }
9
});

1
const changeStream = collection.watch()
2
changeStream.on('change', (event) => {
3
// your logic
4
})

1
// socketHandler.ts
2
const socketHandler = (io: Server) => {
3
  io.on('connection', (socket: any) => {
4
    console.log('A user connected');
5
    socket.on('disconnect', () => {
6
      console.log('A user disconnected');
7
    });
8
  });
9
};
10
export default socketHandler;

1
// server.ts
2
import app from './app'; // Express app
3
import http from 'http';
4
import { Server } from 'socket.io';
5
const server = http.createServer(app);
6
const io = new Server(server);
7
server.listen(5000, async () => {
8
  try {
9
    socketHandler(io);
10
  } catch (error) {
11
    console.error(error);
12
  }
13
});

1
socket.on("UPDATE_DA_LOCATION", async (data) => {
2
  const { email, location } = data;
3
  await collection.findOneAndUpdate({ email }, { $set: { currentLocation: location } });
4
});

1
const watcher = async (io: Server) => {\
2
  const collection = await DeliveryAssociateCollection();\
3
  const changeStream = collection.watch([], { fullDocument: 'updateLookup' });\
4
  changeStream.on('change', (event) => {\
5
    if (event.operationType === 'update') {\
6
        const fullDocument = event.fullDocument;\
7
        io.to(String(fullDocument._id)).emit("DA_LOCATION_CHANGED", fullDocument);\
8
}});};

1
// server.ts
2
import deliveryAssociateWatchers from './watchers/deliveryAssociates';
3
server.listen(5000, async () => {
4
  try {
5
    await dbClient();
6
    socketHandler(io);
7
    await deliveryAssociateWatchers(io);
8
  } catch (error) {
9
    console.error(error);
10
  }
11
});

1
// Driver Simulator
2
const [position, setPosition] = useState(initProps.position);
3
const gpsUpdate = (position) => {
4
  const data = {
5
    email,
6
    location: { type: 'Point', coordinates: [position.lng, position.lat] },
7
  };
8
  socket.emit("UPDATE_DA_LOCATION", data);
9
};
10
useEffect(() => {
11
gpsUpdate(position);
12
}, [position]);
13
return (
14
<MapContainer>
15
<DraggableMarker position={position}/>
16
</MapContainer>
17
)

1
socket.on('SHIPMENT_UPDATED', (data) => {
2
  if (data.deliveryAssociateId) {
3
    const deliveryAssociateId = data.deliveryAssociateId;
4
    socket.emit('SUBSCRIBE_TO_DA', { deliveryAssociateId });
5
  }
6
});

1
const [driverPosition, setDriverPosition] = useState(initProps.position);
2
socket.on('DA_LOCATION_CHANGED', (data) => {
3
  const location = data.location;
4
  setDriverPosition(location);
5
});
6
return (
7
<MapContainer>
8
<Marker position={driverPosition}/>
9
</MapContainer>
10
)