RxPy - 将Live Twitter流转为Rx Observable？

Question

我沿着这个great tutorial跟着使用tweepy在Python中利用实时Twitter流。这将在提供RxJava，RxPy，RxScala或ReactiveX的现场时间内打印Tweets。

from tweepy.streaming import StreamListener 
from tweepy import OAuthHandler 
from tweepy import Stream 
from rx import Observable, Observer 

#Variables that contains the user credentials to access Twitter API 
access_token = "CONFIDENTIAL" 
access_token_secret = "CONFIDENTIAL" 
consumer_key = "CONFIDENTIAL" 
consumer_secret = "CONFIDENTIAL" 


#This is a basic listener that just prints received tweets to stdout. 
class TweetObserver(StreamListener): 

    def on_data(self, data): 
     print(data) 
     return True 

    def on_error(self, status): 
     print(status) 



if __name__ == '__main__': 

    #This handles Twitter authetification and the connection to Twitter Streaming API 
    l = TweetObserver() 
    auth = OAuthHandler(consumer_key, consumer_secret) 
    auth.set_access_token(access_token, access_token_secret) 
    stream = Stream(auth, l) 

    #This line filter Twitter Streams to capture data by the keywords: 'python', 'javascript', 'ruby' 
    stream.filter(track=['rxjava','rxpy','reactivex','rxscala']) 

这是最合适的人选变成通过RxPy一个ReactiveX观察到。但是，我究竟如何将其转化为热门话题Observable？我似乎无法就如何执行Observable.create()地方找到文档...

Answer 1

我前一段时间想通了这一点。你必须定义一个函数来处理一个通过的参数Observer。然后你将它传递给Observable.create()。

from tweepy.streaming import StreamListener 
from tweepy import OAuthHandler 
from tweepy import Stream 
import json 
from rx import Observable 

# Variables that contains the user credentials to access Twitter API 
access_token = "PUT YOURS HERE" 
access_token_secret = "PUT YOURS HERE" 
consumer_key = "PUT YOURS HERE" 
consumer_secret = "PUT YOURS HERE" 


def tweets_for(topics): 
    def observe_tweets(observer): 
     class TweetListener(StreamListener): 
      def on_data(self, data): 
       observer.on_next(data) 
       return True 

      def on_error(self, status): 
       observer.on_error(status) 

     # This handles Twitter authetification and the connection to Twitter Streaming API 
     l = TweetListener() 
     auth = OAuthHandler(consumer_key, consumer_secret) 
     auth.set_access_token(access_token, access_token_secret) 
     stream = Stream(auth, l) 
     stream.filter(track=topics) 

    return Observable.create(observe_tweets).share() 


topics = ['Britain', 'France'] 

tweets_for(topics) \ 
    .map(lambda d: json.loads(d)) \ 
    .subscribe(on_next=lambda s: print(s), on_error=lambda e: print(e))