RxPy - 실시간 트위터 스트림을 수신 가능 상태로 변환합니까?

Question

나는이 great tutorial을 따라 tweepy를 사용하여 파이썬에서 라이브 트위터 스트림을 활용했습니다. RxJava, RxPy, RxScala 또는 ReactiveX가 언급 된 실시간 트윗을 인쇄합니다.

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))