Observable.takeLast;

doc/README.md

@@ -63,6 +63,7 @@ RxLua
   - [sum](#sum)
   - [switch](#switch)
   - [take](#taken)
+  - [takeLast](#takelastcount)
   - [takeUntil](#takeuntilother)
   - [takeWhile](#takewhilepredicate)
   - [tap](#taponnext-onerror-oncompleted)
@@ -639,6 +640,16 @@ Returns a new Observable that only produces the first n results of the original.
 
 ---
 
+#### `:takeLast(count)`
+
+Returns an Observable that produces a specified number of elements from the end of a source Observable.
+
+| Name | Type | Default | Description |
+|------|------|---------|-------------|
+| `count` | number |  | The number of elements to produce. |
+
+---
+
 #### `:takeUntil(other)`
 
 Returns a new Observable that completes when the specified Observable fires.

rx.lua

@@ -1336,6 +1336,36 @@ function Observable:take(n)
   end)
 end
 
+--- Returns an Observable that produces a specified number of elements from the end of a source
+-- Observable.
+-- @arg {number} count - The number of elements to produce.
+-- @returns {Observable}
+function Observable:takeLast(count)
+  return Observable.create(function(observer)
+    local buffer = {}
+
+    local function onNext(...)
+      table.insert(buffer, util.pack(...))
+      if #buffer > count then
+        table.remove(buffer, 1)
+      end
+    end
+
+    local function onError(message)
+      return observer:onError(message)
+    end
+
+    local function onCompleted()
+      for i = 1, #buffer do
+        observer:onNext(util.unpack(buffer[i]))
+      end
+      return observer:onCompleted()
+    end
+
+    return self:subscribe(onNext, onError, onCompleted)
+  end)
+end
+
 --- Returns a new Observable that completes when the specified Observable fires.
 -- @arg {Observable} other - The Observable that triggers completion of the original.
 -- @returns {Observable}

src/operators/takeLast.lua

@@ -0,0 +1,32 @@
+local Observable = require 'observable'
+local util = require 'util'
+
+--- Returns an Observable that produces a specified number of elements from the end of a source
+-- Observable.
+-- @arg {number} count - The number of elements to produce.
+-- @returns {Observable}
+function Observable:takeLast(count)
+  return Observable.create(function(observer)
+    local buffer = {}
+
+    local function onNext(...)
+      table.insert(buffer, util.pack(...))
+      if #buffer > count then
+        table.remove(buffer, 1)
+      end
+    end
+
+    local function onError(message)
+      return observer:onError(message)
+    end
+
+    local function onCompleted()
+      for i = 1, #buffer do
+        observer:onNext(util.unpack(buffer[i]))
+      end
+      return observer:onCompleted()
+    end
+
+    return self:subscribe(onNext, onError, onCompleted)
+  end)
+end

tests/observable.lua

@@ -269,6 +269,7 @@ describe('Observable', function()
   dofile('tests/sum.lua')
   dofile('tests/switch.lua')
   dofile('tests/take.lua')
+  dofile('tests/takeLast.lua')
   dofile('tests/takeUntil.lua')
   dofile('tests/takeWhile.lua')
   dofile('tests/tap.lua')

tests/takeLast.lua

@@ -0,0 +1,28 @@
+describe('takeLast', function()
+  it('produces an error if its parent errors', function()
+    expect(Rx.Observable.throw():takeLast(1).subscribe).to.fail()
+  end)
+
+  it('produces an error if the count is not specified', function()
+    expect(Rx.Observable.fromRange(3):takeLast().subscribe).to.fail()
+  end)
+
+  it('produces nothing if the count is zero', function()
+    local observable = Rx.Observable.fromRange(3):takeLast(0)
+    expect(observable).to.produce.nothing()
+  end)
+
+  it('produces all values if the count is greater than the number of elements produced', function()
+    local observable = Rx.Observable.fromRange(3):takeLast(10)
+    expect(observable).to.produce(1, 2, 3)
+  end)
+
+  it('produces no elements if the count is less than or equal to zero', function()
+    expect(Rx.Observable.fromRange(3):takeLast(0)).to.produce.nothing()
+    expect(Rx.Observable.fromRange(3):takeLast(-1)).to.produce.nothing()
+  end)
+
+  it('produces no elements if the source Observable produces no elements', function()
+    expect(Rx.Observable.empty():takeLast(1)).to.produce.nothing()
+  end)
+end)

tools/concat.lua

@@ -47,6 +47,7 @@ local files = {
   'src/operators/sum.lua',
   'src/operators/switch.lua',
   'src/operators/take.lua',
+  'src/operators/takeLast.lua',
   'src/operators/takeUntil.lua',
   'src/operators/takeWhile.lua',
   'src/operators/tap.lua',