c# - 在 Azure SQL 数据库中插入/更新/删除新记录时将更新推送到 SignalR 流中

Question

我正在使用 SignalR 开发实时数据网格。我有一个 SignalR 集线器。

• 客户端发送 GetStocks向服务器发送消息。
• 服务器响应初始项目列表。
• 客户订阅 GetStockTickerStream溪流。该流仅用于更新。

现在我想用实际数据替换随机生成的股票行情数据，并且当在数据库中插入/更新/删除项目时，将该更新推送到 SignalR GetStockTickerStream .

回顾一下，有没有办法在 Azure SQL 数据库上放置触发器，以便在数据库中插入/更新/删除项目时触发 Azure 函数？

关于如何做到这一点的一个很好的解释和一个小片段会很棒。

public sealed class StockTickerHub : Hub
{
    private readonly IStockTickerService _stockTickerService;

    public StockTickerHub(IStockTickerService stockTickerService)
    {
        _stockTickerService = stockTickerService;
    }

    public ChannelReader<Stock> GetStockTickerStream(CancellationToken cancellationToken)
    {
        var channel = Channel.CreateUnbounded<Stock>();
        
        _ = _stockTickerService.SubscribeAsync(channel.Writer, cancellationToken);
        
        return channel.Reader;
    }

    public async Task GetStocks()
    {
        var stocks = await _stockTickerService.GetStocksAsync();
        await Clients.Caller.SendAsync("ReceiveStocks", stocks);
    }
}

public class Stock
{
    public required long Id { get; init; }

    public required string Symbol { get; init; }
    
    public required double Price { get; init; }
}

public interface IStockTickerService
{
    Task<IEnumerable<Stock>> GetStocksAsync();
    
    Task SubscribeAsync(ChannelWriter<Stock> writer, CancellationToken cancellationToken);
}

public sealed class StockTickerService : IStockTickerService
{
    private static readonly string[] Stocks =
    {
        "TESLA", "S&P 500", "DAX", "NASDAQ", "DOW"
    };

    public Task<IEnumerable<Stock>> GetStocksAsync()
    {
        return Task.FromResult(Enumerable.Range(1, 10).Select(index => new Stock
        {
            Id = index,
            Symbol = Stocks[Random.Shared.Next(Stocks.Length)],
            Price = Math.Round(Random.Shared.NextDouble() * 100, 2)
        }));
    }
    
    public async Task SubscribeAsync(ChannelWriter<Stock> writer, CancellationToken cancellationToken)
    {
        try
        {
            while (true)
            {
                var stock = new Stock
                {
                    Id = 1,
                    Symbol = "TESLA",
                    Price = Math.Round(Random.Shared.NextDouble() * 100, 2)
                };
                await writer.WriteAsync(stock, cancellationToken);

                await Task.Delay(1000, cancellationToken);
            }
        }
        finally
        {
            writer.Complete();
        }
    }
}

public static class ServiceCollectionExtensions
{
    public static IServiceCollection AddDataGrid(this IServiceCollection services)
    {
        ArgumentNullException.ThrowIfNull(services);

        services.AddSingleton<IStockTickerService, StockTickerService>();
        
        return services;
    }

    public static IEndpointRouteBuilder UseDataGrid(this IEndpointRouteBuilder app)
    {
        ArgumentNullException.ThrowIfNull(app);
        
        app.MapHub<StockTickerHub>("/stockticker");
        
        return app;
    }
}

import { useEffect } from "react";
import { useDispatch, useSelector } from "react-redux";
import { Table } from "antd";
import { HubConnectionState } from "redux-signalr";
import hubConnection from "../store/middlewares/signalr/signalrSlice";
import { Stock, addStock } from "../store/reducers/stockSlice";
import { RootState } from "../store";
import "./DataGrid.css";

const DataGrid = () => {
  const dispatch = useDispatch();
  const stocks = useSelector((state: RootState) => state.stock.stocks);

  useEffect(() => {
    if (hubConnection.state !== HubConnectionState.Connected) {
      hubConnection
        .start()
        .then(() => {
          console.log("Started connection via SignalR");

          hubConnection.send("GetStocks");

          hubConnection.stream("GetStockTickerStream").subscribe({
            next: async (item: Stock) => {
              console.log(item);
              dispatch(addStock(item)); // Dispatch addStock action to update Redux store
            },
            complete: () => {
              console.log("Completed");
            },
            error: (err) => {
              console.error(err);
            },
          });
        })
        .catch((err) => console.error(`Faulted: ${err.toString()}`));
    }

    // return () => {
    //   hubConnection
    //     .stop()
    //     .then(() => {})
    //     .catch((err) => console.error(err.toString()));
    // };
  }, [dispatch]);

  return (
    <Table dataSource={stocks} rowKey={(record) => record.id}>
      <Table.Column title="ID" dataIndex="id" key="id" />
      <Table.Column title="Symbol" dataIndex="symbol" key="symbol" />
      <Table.Column title="Price" dataIndex="price" key="price" />
    </Table>
  );
};

export default DataGrid;

import { createSlice, PayloadAction } from "@reduxjs/toolkit";

export type Stock = Readonly<{
  id: number;
  symbol: string;
  price: number;
}>;

export type StockState = Readonly<{
  stocks: Stock[];
}>;

const initialState: StockState = {
  stocks: [],
};

const stockSlice = createSlice({
  name: "stock",
  initialState: initialState,
  reducers: {
    getStocks: (state, action: PayloadAction<Stock[]>) => {
      state.stocks = [...state.stocks, ...action.payload];
    },
    addStock: (state, action: PayloadAction<Stock>) => {
      const stockIndex = state.stocks.findIndex(
        (stock) => stock.id === action.payload.id
      );
    
      if (stockIndex !== -1) {
        // If the stock already exists, update its price
        state.stocks[stockIndex].price = action.payload.price;
      } else {
        // If the stock doesn't exist, add it to the array
        state.stocks.push(action.payload);
      }
    },
  },
});

export const { getStocks, addStock } = stockSlice.actions;

export default stockSlice.reducer;

Answer 1

经过测试，我发现azure sql数据库不支持SqlDependency OnChange事件。如果你想了解更多关于SqlDependency的信息，可以查看this thread .

但我找到了 Azure SQL 触发器。 Azure SQL 触发器绑定(bind)使用轮询循环来检查更改，在检测到更改时触发用户函数。

Azure SQL bindings for Azure Functions overview (preview)