Changeset 430 for trunk/lib/Perlbal/Plugin/AtomStream.pm

Timestamp:

08/25/05 00:21:24 (4 years ago)

Author:

bradfitz

Message:

resume functionality

Files:

• trunk/lib/Perlbal/Plugin/AtomStream.pm (modified) (7 diffs)

trunk/lib/Perlbal/Plugin/AtomStream.pm

@@ -9 +9 @@
 use warnings;
 
-our @subs;  # subscribers
+our @subs;    # subscribers
+our @recent;  # recent items in format [$epoch, $atom_ref]
+
+our $last_timestamp = 0;
 
 use constant MAX_LAG => 262144;
@@ -16 +19 @@
     my $class = shift;
     my $atomref = shift;
+
+    # maintain queue of last 60 seconds worth of posts
+    my $now = time();
+    push @recent, [ $now, $atomref ];
+    shift @recent while @recent && $recent[0][0] <= $now - 60;
+
+    emit_timestamp($now) if $now > $last_timestamp;
 
     my $need_clean = 0;
@@ -33 +43 @@
                 $s->write(\ "<sorryTooSlow youMissed=\"$skip_count\" />\n");
             }
-            $s->write($atomref);
+            $s->watch_write(0) if $s->write($atomref);
         }
     }
@@ -39 +49 @@
     if ($need_clean) {
         @subs = grep { ! $_->{closed} } @subs;
+    }
+}
+
+sub emit_timestamp {
+    my $time = shift;
+    $last_timestamp = $time;
+    foreach my $s (@subs) {
+        next if $s->{closed};
+        $s->{alive_time} = $time;
+        $s->write(\ "<time>$time</time>\n");
     }
 }
@@ -48 +68 @@
     Perlbal::Socket::register_callback(1, sub {
         my $now = time();
-        foreach my $s (@subs) {
-            next if $s->{closed};
-            $s->{alive_time} = $now;
-            $s->write(\ "<time>$now</time>\n");
-        }
+        emit_timestamp($now) if $now > $last_timestamp;
         return 1;
     });
@@ -61 +77 @@
         return 0 unless $hds;
         my $uri = $hds->request_uri;
-        return 0 unless $uri =~ m!^/atom-stream\.xml$!;
+        return 0 unless $uri =~ m!^/atom-stream\.xml(?:\?since=(\d+))?$!;
+        my $since = $1 || 0;
 
         my $res = $self->{res_headers} = Perlbal::HTTPHeaders->new_response(200);
@@ -70 +87 @@
 
         $self->write($res->to_string_ref);
-        $self->write(\ "<?xml version='1.0' encoding='utf-8' ?>\n<atomStream>\n");
+
+        my $last_rv = $self->write(\ "<?xml version='1.0' encoding='utf-8' ?>\n<atomStream><!-- since=$since -->\n");
+
+        # if they'd like a playback, give them all items >= time requested
+        if ($since) {
+            foreach my $item (@recent) {
+                next if $item->[0] < $since;
+                $last_rv = $self->write($item->[1]);
+            }
+        }
+
+        $self->watch_write(0) if $last_rv;
         return 1;
     });