Start subjects;

examples/subject.lua

@@ -0,0 +1,15 @@
+local Rx = require 'rx'
+
+local subject = Rx.Subject.create()
+
+subject:subscribe(function(x)
+  print('observer a ' .. x)
+end)
+
+subject:subscribe(function(x)
+  print('observer b ' .. x)
+end)
+
+subject:onNext(1)
+subject:onNext(2)
+subject:onNext(3)

rx.lua

@@ -369,11 +369,33 @@ end
 
 Scheduler.Cooperative = Cooperative
 
+local Subject = setmetatable({}, Observable)
+Subject.__index = Subject
+
+function Subject.create(initialValue)
+  local self = {
+    observers = {}
+  }
+
+  return setmetatable(self, Subject)
+end
+
+function Subject:subscribe(onNext, onError, onComplete)
+  table.insert(self.observers, Observer.create(onNext, onError, onComplete))
+end
+
+function Subject:onNext(value)
+  for i = 1, #self.observers do
+    self.observers[i]:onNext(value)
+  end
+end
+
 rx = {
   Observer = Observer,
   Observable = Observable,
   Scheduler = Scheduler,
-  scheduler = Scheduler.Cooperative.create()
+  scheduler = Scheduler.Cooperative.create(),
+  Subject = Subject
 }
 
 return rx