Zbuduj swoją pierwszą aplikację

import asyncio
import os
from datetime import datetime, timezone
from ironflock import IronFlock
 
async def main():
    # Zarejestruj to urządzenie jako online
    await flock.append_to_table("devices", {
        "device_key": os.environ.get('DEVICE_KEY'),
        "tsp": datetime.now(timezone.utc).isoformat(),
        "status": "online",
        "device_name": os.environ.get('DEVICE_NAME'),
        "deleted": False
    })
 
    # ... reszta twojej logiki
 
flock = IronFlock(mainFunc=main)
flock.run()

import { IronFlock } from "ironflock";
 
const flock = new IronFlock();
await flock.start();
 
// Zarejestruj to urządzenie jako online
await flock.appendToTable("devices", [{
    device_key: process.env.DEVICE_KEY,
    tsp: new Date().toISOString(),
    status: "online",
    device_name: process.env.DEVICE_NAME,
    deleted: false
}]);

package main
 
import (
	"context"
	"log"
	"time"
 
	ironflock "github.com/RecordEvolution/ironflock-go"
)
 
func main() {
	ifl, err := ironflock.New()
	if err != nil {
		log.Fatal(err)
	}
 
	// Run łączy się, uruchamia funkcję i kończy działanie po jej powrocie albo przy SIGINT lub SIGTERM.
	err = ifl.Run(context.Background(), func(ctx context.Context) error {
		// Zarejestruj to urządzenie jako online
		_, err := ifl.AppendToTable(ctx, "devices", ironflock.Row{
			"device_key":  ifl.DeviceKey(), // z DEVICE_KEY
			"tsp":         time.Now(),
			"status":      "online",
			"device_name": ifl.DeviceName(), // z DEVICE_NAME
			"deleted":     false,
		})
		if err != nil {
			return err
		}
 
		// ... reszta twojej logiki
 
		<-ctx.Done() // działaj, dopóki aplikacja nie zostanie zatrzymana
		return nil
	})
	if err != nil {
		log.Fatal(err)
	}
}

async def main():
    # ... rejestracja urządzenia ...
 
    # Odczytaj istniejące serwery Modbus
    existing = await flock.getHistory("modbus_servers", {
        "limit": 1000,
        "filterAnd": [
            # "latest": True zwraca tylko najnowszy wiersz dla każdego server_id
            {"latest": True},
            # deleted IS NULL OR deleted = false
            {"combinator": "OR", "filters": [
                {"column": "deleted", "operator": "IS NULL"},
                {"column": "deleted", "operator": "=", "value": False}
            ]}
        ]
    })
    for server in existing:
        create_modbus_connection(server)
 
    # Reaguj na nowe serwery w czasie rzeczywistym
    def on_server_change(*args, **kwargs):
        server = args[0] if args else kwargs
        if server.get("deleted"):
            delete_modbus_connection(server["server_id"])
        else:
            create_modbus_connection(server)
 
    await flock.subscribe_to_table("modbus_servers", on_server_change)

// Odczytaj istniejące serwery Modbus
const existing = await flock.getHistory("modbus_servers", {
    limit: 1000,
    filterAnd: [
        // latest: true zwraca tylko najnowszy wiersz dla każdego server_id
        { latest: true },
        // deleted IS NULL OR deleted = false
        { combinator: "OR", filters: [
            { column: "deleted", operator: "IS NULL" },
            { column: "deleted", operator: "=", value: false }
        ] }
    ]
});
existing.forEach(server => createModbusConnection(server));
 
// Reaguj na nowe serwery w czasie rzeczywistym
await flock.subscribeToTable("modbus_servers", (args) => {
    const server = args[0];
    if (server.deleted) {
        deleteModbusConnection(server.server_id);
    } else {
        createModbusConnection(server);
    }
});

// Wewnątrz funkcji przekazanej do ifl.Run:
// ... rejestracja urządzenia ...
 
// Odczytaj istniejące serwery Modbus
existing, err := ifl.GetHistory(ctx, "modbus_servers", &ironflock.TableQueryParams{
	Limit: 1000,
	FilterAnd: []ironflock.Filter{
		// Latest() zwraca tylko najnowszy wiersz dla każdego server_id
		ironflock.Latest(),
		// deleted IS NULL OR deleted = false
		ironflock.Or(
			ironflock.IsNull("deleted"),
			ironflock.Where("deleted", "=", false),
		),
	},
})
if err != nil {
	return err
}
for _, server := range existing {
	createModbusConnection(ctx, server)
}
 
// Reaguj na nowe serwery w czasie rzeczywistym
_, err = ifl.SubscribeToTable(ctx, "modbus_servers", func(ev *ironflock.Event) {
	server := ev.Row()
	if deleted, _ := server["deleted"].(bool); deleted {
		serverID, _ := server["server_id"].(string)
		deleteModbusConnection(serverID)
	} else {
		createModbusConnection(ctx, server)
	}
})
if err != nil {
	return err
}

async def modbus_poll_loop(server_id, client):
    while True:
        values = client.read_registers()
        tsp = datetime.now(timezone.utc).isoformat()
        for metric, value in values.items():
            await flock.publish_to_table("readings", {
                "tsp": tsp,
                "server_id": server_id,
                "metric": metric,
                "value": value
            })
        await asyncio.sleep(5)

async function modbusPollingLoop(serverId, client) {
    setInterval(async () => {
        const values = client.readRegisters();
        const tsp = new Date().toISOString();
        for (const [metric, value] of Object.entries(values)) {
            await flock.publishToTable("readings", [{
                tsp,
                server_id: serverId,
                metric,
                value
            }]);
        }
    }, 5000);
}

// createModbusConnection uruchamia jedną pętlę na serwer: go modbusPollLoop(ctx, ifl, serverID, client)
func modbusPollLoop(ctx context.Context, ifl *ironflock.IronFlock, serverID string, client *ModbusClient) {
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()
	for {
		values, err := client.ReadRegisters() // map[string]float64: nazwa metryki na wartość
		if err != nil {
			log.Print(err)
		}
		tsp := time.Now()
		for metric, value := range values {
			err := ifl.PublishToTable(ctx, "readings", ironflock.Row{
				"tsp":       tsp,
				"server_id": serverID,
				"metric":    metric,
				"value":     value,
			})
			if err != nil {
				log.Print(err)
			}
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}