BehaviorSubject mistakenly refers to check_unsubscribed as a class method

Author: Bittrance

lib/rx/subjects/behavior_subject.rb

@@ -31,8 +31,8 @@ def has_observers?
 
     # Gets the current value or throws an exception.
     def value
-      gate.synchronize do 
-        self.check_unsubscribed
+      gate.synchronize do
+        check_unsubscribed
         raise @error if @error
         @value
       end
@@ -41,8 +41,8 @@ def value
     # Notifies all subscribed observers about the end of the sequence.
     def on_completed
       os = nil
-      @gate.synchronize do 
-        self.check_unsubscribed
+      @gate.synchronize do
+        check_unsubscribed
 
         unless @stopped
           os = @observers.clone
@@ -60,7 +60,7 @@ def on_error(error)
 
       os = nil
       @gate.synchronize do
-        self.check_unsubscribed
+        check_unsubscribed
 
         unless @stopped
           os = @observers.clone
@@ -77,7 +77,7 @@ def on_error(error)
     def on_next(value) 
       os = nil
       @gate.synchronize do
-        self.check_unsubscribed
+        check_unsubscribed
         @value = value
         os = @observers.clone unless @stopped
       end
@@ -91,7 +91,7 @@ def subscribe(observer)
 
       err = nil
       gate.synchronize do
-        self.check_unsubscribed
+        check_unsubscribed
 
         unless @stopped
           observers.push(observer)

test/rx/subjects/test_behavior_subject.rb

@@ -0,0 +1,31 @@
+# Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
+
+require 'test_helper'
+
+class TestBehaviorSubject < Minitest::Test
+  def test_subscriber_notified_on_change
+    value = 0
+    subject = Rx::BehaviorSubject.new 0
+    subject.as_observable.subscribe { |update| value = update }
+    subject.on_next 1
+    assert_equal 1, value
+  end
+
+  def test_multiple_observers_notified_on_change
+    value1 = 0
+    value2 = 0
+    subject = Rx::BehaviorSubject.new 0
+    subject.as_observable.subscribe { |update| value1 = update }
+    subject.as_observable.subscribe { |update| value2 = update }
+    subject.on_next 1
+    assert_equal 1, value1
+    assert_equal 1, value2
+  end
+
+  def test_errors_on_next_when_unsubscribed
+    subject = Rx::BehaviorSubject.new 0
+    subject.as_observable.subscribe { }
+    subject.unsubscribe
+    assert_raises(RuntimeError) { subject.on_next 1 }
+  end
+end