view release on metacpan or  search on metacpan

lib/Stardust.pm  view on Meta::CPAN

        }
      }
    },
  ),

  # Message - [public]
  # This controller emits a stream of messages to long-polling clients.
  C(
    Message => [ '/channel/([\w+]+)/stream/([.\d]+)' ],
    get => sub {
      warn "coro [$Coro::current]" if $CONFIG{debug};
      my ($self, $channels, $client_id) = @_;
      my $input  = $self->input;
      my $cr     = $self->cr;
      my @ch     = split(/\+/, $channels);
      my $last   = time;
      while (1) {
        # Output
        warn "top of loop" if $CONFIG{debug};
        my @messages = 
          grep { defined } 

lib/Stardust.pm  view on Meta::CPAN

          warn "printing...".encode_json(\@messages) if $CONFIG{debug};
          $cr->print(encode_json(\@messages));
        };
        $x->join;
        $last = time;

        # Hold for a brief moment until the next long poll request comes in.
        warn "waiting for next request" if ($CONFIG{debug});
        $cr->next;

        # Start 1 coro for each channel we're listening to.
        # Each coro will have the same Coro::Signal object, $activity.
        my $activity = Coro::Signal->new;
        my @coros = map {
          my $ch = channel($_);
          async { $ch->signal->wait; $activity->broadcast };
        } @ch;

        # When running this behind a reverse proxy,
        # it's useful to timeout before your proxy kills the connection.
        push @coros, async {
          my $timeout = Coro::Timer::timeout $CONFIG{timeout};
          while (not $timeout) {
            Coro::schedule;
          }
          warn "timeout\n" if $CONFIG{debug};
          $activity->broadcast;
        };

        # The first coro that does $activity->broadcast wins.
        warn "waiting for activity on any of (@ch); last is $last" if $CONFIG{debug};
        $activity->wait;

        # Cancel the remaining coros.
        for (@coros) { $_->cancel }
      }
    },
    continuity => 1,
  ),

);

1;

__END__