Symfoware

Apache Thrift クライアントとサーバー間の通信にunix socketを使用する

Thriftを使用し、PythonとNode.jsそれぞれでバイナリデータを送受信するプログラムを作成しました。
Apache Thrift Pythonでバイナリデータ(画像)の送受信
Apache Thrift Node.jsでバイナリデータ(画像)の送受信

クライアントとサーバー間の通信にはポート番号を指定してソケットを使用しましたが、
この部分をユニックスソケットに変更してみます。

Python

TSocket.pyのソースを見てみると
https://github.com/apache/thrift/blob/master/lib/py/src/transport/TSocket.py
host、portではなく、unix_socketで指定すれば良いようです。

ソケットは書き込み権限があるパスならどこでも良いのですが、
今回は「/tmp/thrift.sock」としました。
ソースを修正してみます。

・client.py

# Thriftコンパイラーで出力したテンプレートをパスに追加
import sys
sys.path.append('gen-py')
# テンプレートのファイル名に使用した「sample」にある「Transfer」サービスをimport
from sample import Transfer
from thrift import Thrift
from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol
def main():
    # データのやり取りはソケット通信を使用
    # Make socket
    # transport = TSocket.TSocket('localhost', 9090)
    # unix socket使用
    transport = TSocket.TSocket(unix_socket='/tmp/thrift.sock')
    # Buffering is critical. Raw sockets are very slow
    transport = TTransport.TBufferedTransport(transport)
    # プロトコルの指定
    protocol = TBinaryProtocol.TBinaryProtocol(transport)
    # Transferサービスを使用するクライアントを生成
    client = Transfer.Client(protocol)
    # コネクションオープン
    transport.open()
    # 送信するファイル読み込み
    body = open('lena.png', 'rb').read()
    # Transferサービスのsendを実行しファイル送信
    result = client.send('send.png', body)
    print(result)
    # コネクションクローズ
    transport.close()
if __name__ == '__main__':
    main()

サーバー側もunix_socketを使用するよう変更。

・server.py

# Thriftコンパイラーで出力したテンプレートをパスに追加
import sys
sys.path.append('gen-py')
# テンプレートのファイル名に使用した「sample」にある「Transfer」サービスをimport
from sample import Transfer
from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol
from thrift.server import TServer
class TransferHandler:
    def __init__(self):
        self.log = {}
    def send(self, name, body):
        print(name)
        print(len(body))
        # 受信したファイル名で内容を保存
        open(name, 'wb').write(body)
        return 'ok'
if __name__ == '__main__':
    handler = TransferHandler()
    processor = Transfer.Processor(handler)
    #transport = TSocket.TServerSocket(host='127.0.0.1', port=9090)
    # unix socket使用
    transport = TSocket.TServerSocket(unix_socket='/tmp/thrift.sock')
    tfactory = TTransport.TBufferedTransportFactory()
    pfactory = TBinaryProtocol.TBinaryProtocolFactory()
    server = TServer.TSimpleServer(processor, transport, tfactory, pfactory)
    # You could do one of these for a multithreaded server
    # server = TServer.TThreadedServer(
    #     processor, transport, tfactory, pfactory)
    # server = TServer.TThreadPoolServer(
    #     processor, transport, tfactory, pfactory)
    print('Starting the server...')
    server.serve()
    print('done.')

サーバー側を起動。


/tmp/thrift.sockが生成されました。
クライアント側を実行。


TCPソケットと使用したときと同様、画像ファイルの転送が確認できました。

Node.js

Node.jsでもunix socketを使用するよう修正してみます。
connection.jsのソースを見てみると
https://github.com/apache/thrift/blob/master/lib/nodejs/lib/thrift/connection.js

createUDSConnectionを使用すれば良いようです。
ソースを修正します。

・client.js

const fs = require('fs').promises;
// コンパイラで出力したファイルを読み込み
const Transfer = require('./gen-nodejs/Transfer');
const thrift = require('thrift');
const transport = thrift.TBufferedTransport;
const protocol = thrift.TBinaryProtocol;
// unixソケットを使用するよう変更
//const connection = thrift.createConnection('localhost', 9090, {
const connection = thrift.createUDSConnection('/tmp/thrift.sock', {
    transport : transport,
    protocol : protocol
});
connection.on('error', function(err) {
    console.log(err);
});
// データ送信用のクライアント生成
const client = thrift.createClient(Transfer, connection);
async function main() {
    // 送信データ読み込み
    const body = await fs.readFile('lena.png');
    client.send('send.png', body, function(err, response) {
        if (err) {
            console.log(err);
        } else {
            console.log(response);
        }
        connection.end();
    });
}
main();

サーバー側もソースを修正。

・server.js

const fs = require('fs').promises;
// コンパイラで出力したファイルを読み込み
const Transfer = require('./gen-nodejs/Transfer');
const thrift = require('thrift');
const server = thrift.createServer(Transfer, {
    // 送信されたファイルを受信
    send: async function(name, body, result) {
        console.log(name);
        await fs.writeFile(name, body);
        result(null, 'ok');
    }
});
// unixソケットを使用するよう変更
//server.listen(9090);
server.listen('/tmp/thrift.sock');

実行前、/tmp/thrift.sockが残っているとサーバー起動時にエラーとなります。
Python版を実行した後はファイルが残っていたので一旦削除。


サーバー側を起動。


クライアント側を実行してファイルを送信。


画像ファイルの転送が確認できました。

Apache Thrift PythonとNode.js間のバイナリデータ(画像)送受信

Thriftを使用し、PythonとNode.jsそれぞれでバイナリデータを送受信するプログラムを作成しました。
Apache Thrift Pythonでバイナリデータ(画像)の送受信
Apache Thrift Node.jsでバイナリデータ(画像)の送受信

Pythonでデータ送信。Node.jsでデータ復元という異なる言語間での動作を試してみます。

通信定義の作成とコンパイル

前回と同じく「Transfer」というサービス、string(ファイル名)とbinary(ファイルデータ)を
送信する定義を使用します。

・sample.thrift


定義からプログラムの実装を出力します。
Pythonでの実装の出力は「py」
Node.jsの実装は--genには「js:node」を指定して出力を実行。

Python側のクライアントプログラム

Thrift + Python編で使用したのと全く同じプログラムです。

・client.py

# Thriftコンパイラーで出力したテンプレートをパスに追加
import sys
sys.path.append('gen-py')
# テンプレートのファイル名に使用した「sample」にある「Transfer」サービスをimport
from sample import Transfer
from thrift import Thrift
from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol
def main():
    # データのやり取りはソケット通信を使用
    # Make socket
    transport = TSocket.TSocket('localhost', 9090)
    # Buffering is critical. Raw sockets are very slow
    transport = TTransport.TBufferedTransport(transport)
    # プロトコルの指定
    protocol = TBinaryProtocol.TBinaryProtocol(transport)
    # Transferサービスを使用するクライアントを生成
    client = Transfer.Client(protocol)
    # コネクションオープン
    transport.open()
    # 送信するファイル読み込み
    body = open('lena.png', 'rb').read()
    # Transferサービスのsendを実行しファイル送信
    result = client.send('send.png', body)
    print(result)
    # コネクションクローズ
    transport.close()
if __name__ == '__main__':
    main()

Node.js側のサーバープログラム

こちらもThrift + Node.js編で使用したのと同じ内容です。

・server.js

const fs = require('fs').promises;
// コンパイラで出力したファイルを読み込み
const Transfer = require('./gen-nodejs/Transfer');
const thrift = require('thrift');
const server = thrift.createServer(Transfer, {
    // 送信されたファイルを受信
    send: async function(name, body, result) {
        console.log(name);
        await fs.writeFile(name, body);
        result(null, 'ok');
    }
});
server.listen(9090);

実行

まずNode.jsのサーバー側を起動。


続いてPythonのクライアントから送信実行。



サーバー側のディレクトリにsend.pngという名称で画像ファイルが出力されることを確認できました。
Thriftを使用すれば、異なる言語間で簡単にネットワーク越しのデータ連携が実現できますね。
webサーバーも不要ですし便利です。

Apache Thrift Pythonでバイナリデータ(画像)の送受信

Apache Thrift
https://thrift.apache.org/

Thriftは以前に触ったことがありますが、
ThriftをPythonで使用する
当時と状況が変わっているようなので、改めて動かしてみます。

お題として、ファイル名(文字列)とファイルの内容(バイナリ)を送信。
サーバー側でファイルを復元する機能を作成します。

Thrift compilerの取得とビルド

https://thrift.apache.org/download
こちらのダウンロードからThriftのコンパイラー(データ型の定義から実装を出力する)を取得。
Windows用のバイナリ(thrift-0.17.0.exe)も用意されていますね。

Ubuntuへはこちらの手順でインストールしました。
Ubuntu 22.04にApache Thriftコンパイラーをインストール(thrift-compilerとソースからのビルド)

※aptでもインストールできるはずです。

通信定義の作成とコンパイル

データのやり取りを行うサービスやメソッド名、データ型の定義を行います。
定義体の記載方法は、チュートリアルに登場するこちらが参考になります。
shared.thrift
tutorial.thrift

「Transfer」というサービス、string(ファイル名)とbinary(ファイルデータ)を
送信する定義は以下のようになりました。

・sample.thrift


定義からプログラムの実装を出力します。
Pythonの実装が欲しいので--genには「py」を指定して出力を実行。


Windowsの場合


gen-pyというディレクトリが自動的に作成され、ひな形のソースコートが出力されます。



このひな形を利用してプログラム本体を記載していきます。

必要なPythonパッケージのインストール

以前はプログラムに必要な資産全てがコンパイラーから出力されていた気がするのですが、
データ送受信などの共通的なプログラムはpipでインストールする必要がありました。

https://pypi.org/project/thrift/

pipコマンドでインストールします。

プログラムサンプル

チュートリアルを参考にプログラムを作成します。
https://thrift.apache.org/tutorial/py.html

まずクライアント側。
画像ファイルを読み込んでサーバー側に送信します。

送信テストに使用したファイルはこちら。



・client.py

# Thriftコンパイラーで出力したテンプレートをパスに追加
import sys
sys.path.append('gen-py')
# テンプレートのファイル名に使用した「sample」にある「Transfer」サービスをimport
from sample import Transfer
from thrift import Thrift
from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol
def main():
    # データのやり取りはソケット通信を使用
    # Make socket
    transport = TSocket.TSocket('localhost', 9090)
    # Buffering is critical. Raw sockets are very slow
    transport = TTransport.TBufferedTransport(transport)
    # プロトコルの指定
    protocol = TBinaryProtocol.TBinaryProtocol(transport)
    # Transferサービスを使用するクライアントを生成
    client = Transfer.Client(protocol)
    # コネクションオープン
    transport.open()
    # 送信するファイル読み込み
    body = open('lena.png', 'rb').read()
    # Transferサービスのsendを実行しファイル送信
    result = client.send('send.png', body)
    print(result)
    # コネクションクローズ
    transport.close()
if __name__ == '__main__':
    main()

次にサーバー側。
受信したファイル名で内容を保存します。

・server.py

# Thriftコンパイラーで出力したテンプレートをパスに追加
import sys
sys.path.append('gen-py')
# テンプレートのファイル名に使用した「sample」にある「Transfer」サービスをimport
from sample import Transfer
from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol
from thrift.server import TServer
class TransferHandler:
    def __init__(self):
        self.log = {}
    def send(self, name, body):
        print(name)
        print(len(body))
        # 受信したファイル名で内容を保存
        open(name, 'wb').write(body)
        return 'ok'
if __name__ == '__main__':
    handler = TransferHandler()
    processor = Transfer.Processor(handler)
    transport = TSocket.TServerSocket(host='127.0.0.1', port=9090)
    tfactory = TTransport.TBufferedTransportFactory()
    pfactory = TBinaryProtocol.TBinaryProtocolFactory()
    server = TServer.TSimpleServer(processor, transport, tfactory, pfactory)
    # You could do one of these for a multithreaded server
    # server = TServer.TThreadedServer(
    #     processor, transport, tfactory, pfactory)
    # server = TServer.TThreadPoolServer(
    #     processor, transport, tfactory, pfactory)
    print('Starting the server...')
    server.serve()
    print('done.')

サーバー側を起動。



クライアント側のプログラムを実行します。


実行すると、send.pngというファイルが出力されます。
内容も元の画像と同様ですね。
手軽にシステム間でバイナリデータのやり取りが行えて便利です。

Python asyncio並行処理と終了の待ち合わせ、タスク初期値の設定

以前、Pythonのスレッドで並行処理を行う方法を調べました。
Python スレッドによる並行処理と終了の待ち合わせ、スレッド初期値の設定

Python2しか使用できない環境を想定していましたが、
今回はasyncioが使用できる環境を想定し、同様の内容を試してみます。

最も簡単な例

0から4までの数字を出力するプログラムを2つ同時に動かしてみます。

・sample.py

import asyncio
import time
async def funA():
    for i in range(5):
        print('funA:%d' % i)
        await asyncio.sleep(1)
async def funB():
    for i in range(5):
        print('funB:%d' % i)
        await asyncio.sleep(1)
async def main():
    # A,Bそれぞれをタスクとして作成する
    taskA = asyncio.create_task(funA())
    taskB = asyncio.create_task(funB())
    # タスクを実行 並列で実行される
    await taskA
    await taskB
if __name__ == '__main__':
    asyncio.run(main())

非同期実行の関数は「async」をつけて定義する。
asyncio.runで実行。
並行して実行したいものは、asyncio.create_taskで実行をスケジュール。
Taskオブジェクトをawaitして実行終了を待ち合わせる。という感じでしょうか。

実行結果


funcA, funcBの処理はasyncio.create_taskした時点で開始されます。
await taskA, await taskBで各々の処理が終了しているか待ち合わせるイメージです。

async関数への値の設定

実行する関数へ値を渡したい場合はそのまま引数にとればOKです。

・sample.py

import asyncio
import time
async def worker(name):
    for i in range(5):
        # 引数で得た名前と繰り返し回数を表示
        print('Worker%s:%d' % (name, i))
        await asyncio.sleep(1)
async def main():
    # 引数を渡してタスクを作成
    taskA = asyncio.create_task(worker('A'))
    taskB = asyncio.create_task(worker('B'))
    
    # タスクを実行 並列で実行される
    await taskA
    await taskB
    
if __name__ == '__main__':
    asyncio.run(main())

実行結果


スレッドで実行する際はあれこれ工夫しましたが、asyncioだと簡単ですね。

タスク実行結果の値の受け取り

Python スレッドによる並行処理と終了の待ち合わせ、スレッド初期値の設定
こちらの「スレッド終了の待ち合わせ(join)」で試した内容のasyncio版です。
例では、0から引数に与えられた数字までを合算するプログラムを並行実行。
各々の実行結果から得られた値をさらに合算するサンプルでした。


まず、与えられた数値までの合算を行うサンプル。

import asyncio
import time
async def worker(max):
    # 0から与えられた数値まで合算
    total = 0
    for i in range(max):
        total += i
        # 本来必要ないが、処理に時間がかかる例としてwait
        await asyncio.sleep(0.1)
    
    # 結果を取得
    return total
async def main():
    # 引数を渡してタスクを作成
    taskA = asyncio.create_task(worker(10))
    
    # タスクから戻り値を得られる
    totalA = await taskA
    print(totalA)
    
if __name__ == '__main__':
    asyncio.run(main())

await [タスク]の戻り値として関数からの戻り値が得られます。
実行結果



過去サンプルと同様の処理を記載してみます。

・sample.py

import asyncio
import time
async def worker(name, max):
    # 0から与えられた数値まで合算
    total = 0
    for i in range(max):
        print('Worker%s:%d' % (name, i))
        total += i
        # 本来必要ないが、処理に時間がかかる例としてwait
        await asyncio.sleep(0.1)
    
    # 結果を取得
    return total
async def main():
    # 引数を渡してタスクを作成
    taskA = asyncio.create_task(worker('A', 10))
    taskB = asyncio.create_task(worker('B', 20))
    
    # タスクから戻り値を得られる
    totalA = await taskA
    totalB = await taskB
    print(totalA + totalB)
    
if __name__ == '__main__':
    asyncio.run(main())

実行結果


2つの加算処理が並列で実行。
各々の計算結果を受け取り、合算値を表示することができました。

スレッドを使用するよりお手軽に実装できますね。

ISO-2022-JP対応 SMTPメール送信テストサーバー PyMailHog 0.0.4

ISO-2022-JPに対応したMailHogのPythonクローン「PyMailHog」
PyMailHog

cssフレームワークにBootstrap 3系を使用していたのですが、
Picnic CSSを使用するよう変更しました。
これに伴い、jqueryに依存する必要がなくなったので、jqueryを削除しています。

加えて、.eml形式でメールデータをダウンロードする機能を追加。
ローカルのメーラーで表示を確認できるようになりました。

スクリーンショット

使い方

PyMailHogファイルはzipappで実行可能なPython zip書庫としています。
zipapp --- 実行可能な Python zip 書庫を管理する

ダウンロードリンクにあるPyMailHog単体で動作します。
PyMailHog ダウンロード

pythonコマンドに続いてファイル名を指定すれば起動します。


Linuxではダウンロードしたファイルに実行権限を付与
そのまま実行することが可能です。


Ctrl + cキーで終了します。

デフォルトでsmtpポートは1025
httpポートは8025で待ち受けます。

--smtpportオプションでsmtpポート
--httpportオプションでhttpポートを指定可能です。


-hで指定可能なオプションが表示されます。