在通道消费者类中使用 django 信号

Question

我正在尝试开发一种拍卖类型的系统，在该系统中，客户下订单，然后不同的商店可以为该订单提供价格。

该系统的一个有趣部分是，在最初创建订单时，可用商店将有 60 秒的时间来提供各自的报价。当第一家商店提出报价时，“拍卖”现在只有接下来的 20 秒时间让其他商店提出自己的报价。如果他们确实在这个更小的分配时间内提出了另一个报价，那么这 20 秒就会被刷新。只要有足够的时间，就可以继续收到报价，时间不能超过最初给出的 60 秒。

class Order(models.Model):
    customer = models.ForeignKey(Customer)
    create_time = models.DateTimeField(auto_now_add=True)
    update_time = models.DateTimeField(auto_now_add=True)
    total = models.FloatField(default=0)
    status = models.IntegerField(default=0)
    delivery_address = models.ForeignKey(DeliveryAddress)
    store = models.ForeignKey(Store, null=True, blank=True, related_name='orders', on_delete=models.CASCADE)
    credit_card = models.ForeignKey(CreditCard, null=True, blank=True, related_name='orders')

class OrderOffer(models.Model):
    store = models.ForeignKey(Store, related_name="offers", on_delete=models.CASCADE)
    order = models.ForeignKey(Order, related_name="offers", on_delete=models.CASCADE)
    create_time = models.DateTimeField(auto_now_add=True)

除了这些要求之外，我还想在新报价到达时实时更新客户端。为此，我使用了django-channels WebSockets 的实现。

我有以下consumers.pyfile:

from channels.generic.websockets import WebsocketConsumer
from threading import Timer
from api.models import Order, OrderOffer
from django.db.models.signals import post_save
from django.dispatch import receiver

class OrderConsumer(WebsocketConsumer):

    def connect(self, message, **kwargs):
        """
        Initialize objects here.
        """
        order_id = int(kwargs['order_id'])
        self.order = Order.objects.get(id=order_id)
        self.timer = Timer(60, self.sendDone)
        self.timer.start()
        self.message.reply_channel.send({"accept": True})

    def sendDone(self):
        self.send(text="Done")

    # How do I bind self to onOffer?
    @receiver(post_save, sender=OrderOffer)
    def onOffer(self, sender, **kwargs):
        self.send(text="Offer received!")
        if (len(self.offers) == 0):
            self.offerTimer = Timer(20, self.sendDone)
            self.offers = [kwargs['instance'],]
        else:
            self.offerTimer = Timer(20, self.sendDone)

        self.offers.append(kwargs['instance'])


    def receive(self, text=None, bytes=None, **kwargs):
        # Echo
        self.send(text=text, bytes=bytes)

    def disconnect(self, message, **kwargs):
        """
        Perform necessary disconnect operations.
        """
        pass

我已经成功地在我的客户端和服务器之间建立了一个 WebSocket 通信通道。我已经测试了发送消息，一切似乎都很好。现在我想检测新OrderOffer 的创建，并向客户端发送通知。为此，我需要访问self 变量以使用self.send，这是不可能的，因为信号装饰器不发送此参数。我尝试通过用 self 声明 onOffer 来强制它，但我收到以下错误：

TypeError: onOffer() missing 1 required positional argument: 'self'

如果我能以某种方式访问​​信号集的关键字参数，我可能会执行以下操作： context = self.

我将不胜感激任何帮助，甚至是我原来问题的替代解决方案。

Answer 1

如果有人偶然发现，这就是我在signals.py 中解决它的方法。我有一个Job，每次更改时都需要将其status 发送给客户端。这是我的signals.py：

import channels.layers
from asgiref.sync import async_to_sync

from django.db.models.signals import post_save
from django.dispatch import receiver

from .models import Job


def send_message(event):
    '''
    Call back function to send message to the browser
    '''
    message = event['text']
    channel_layer = channels.layers.get_channel_layer()
    # Send message to WebSocket
    async_to_sync(channel_layer.send)(text_data=json.dumps(
        message
    ))


@receiver(post_save, sender=Job, dispatch_uid='update_job_status_listeners')
def update_job_status_listeners(sender, instance, **kwargs):
    '''
    Sends job status to the browser when a Job is modified
    '''

    user = instance.owner
    group_name = 'job-user-{}'.format(user.username)

    message = {
        'job_id': instance.id,
        'title': instance.title,
        'status': instance.status,
        'modified': instance.modified.isoformat(),
    }

    channel_layer = channels.layers.get_channel_layer()

    async_to_sync(channel_layer.group_send)(
        group_name,
        {
            'type': 'send_message',
            'text': message
        }
    )

顺便说一句，我有一个消费者 class JobUserConsumer(AsyncWebsocketConsumer) 我在其中定义了组：

async def connect(self):

    user = self.scope["user"]
    self.group_name = 'job-user-{}'.format(user.username)

    await self.channel_layer.group_add(
        self.group_name,
        self.channel_name
    )

    await self.accept()

我使用的项目在这里：https://github.com/ornl-ndav/django-remote-submission/tree/master/django_remote_submission

Answer 2

对于那些仍然有网络套接字问题的人，这可能会有所帮助：

from api.models import Order, OrderOffer
from asgiref.sync import async_to_sync
import channels.layers
from channels.generic.websocket import JsonWebsocketConsumer
from django.db.models import signals
from django.dispatch import receiver


class OrderOfferConsumer(JsonWebsocketConsumer):
    def connect(self):
        async_to_sync(self.channel_layer.group_add)(
            'order_offer_group',
            self.channel_name
        )
        self.accept()

    def disconnect(self, close_code):
        async_to_sync(self.channel_layer.group_discard)(
            'order_offer_group',
            self.channel_name
        )
        self.close()

    def receive_json(self, content, **kwargs):
        print(f"Received event: {content}")

    def events_alarm(self, event):
        self.send_json(event['data'])

    @staticmethod
    @receiver(signals.post_save, sender=OrderOffer)
    def order_offer_observer(sender, instance, **kwargs):
        layer = channels.layers.get_channel_layer()
        async_to_sync(layer.group_send)('order_offer_group', {
            'type': 'events.alarm',
            'data': {
                'text': 'Offer received',
                'id': instance.pk
            }
        })

在 urls.py 中需要注册一个新的 webscoket 路由：

websocket_urlpatterns = [url(r'^order_offer$', OrderOfferConsumer)]

Answer 3

如果您想从“外部”与消费者对话 - 在这种情况下，通过模型保存方法 - 您需要使用通道层与其对话：http://channels.readthedocs.io/en/latest/topics/channel_layers.html

基本上，您需要：

• 在启动时将消费者添加到组（可能基于其订单 ID）
• 只要有带有自定义 type 的新 OrderOffer，就向群组发送消息 - 例如{"type": "order.new_offer", "order_offer_id": 45}
• 在处理此问题的消费者上定义一个处理程序 - 它与类型名称匹配，因此在本例中为def order_new_offer(self, event):
• 然后，在该处理程序中，您可以使用self.send 与套接字通话（如果您需要额外的信息来发送给您未放入事件消息中的客户端，则可以查询数据库）。

您可以在 MultiChat 示例项目中看到它的变体：https://github.com/andrewgodwin/channels-examples/tree/master/multichat