واٹرمل پبلشر کمپونینٹ (میسج ٹرانزیکشن)

ٹرانزیکشن میں پیغامات شائع کرنے کی اہمیت

وقت کے ساتھ واقعے کے مبنی ایپلیکیشنز کا سامنا کرتے ہوئے، ایسی صورتحال پیدا ہوسکتی ہے جب آپ کو درختورد حالت کو قائم رکھنا ہوگا اور دوسرے حصوں کو بتانے کی ضرورت ہوگی کہ بلحاظ واقعہ کیا ہوا ہے۔ ایک مثالی صورتحال میں، آپ کو درختورد حالت کو قائم رکھنا ہوگا اور ایک پیغام شائع کرنا ایک ہی ٹرانزیکشن کے اندر، کیونکہ اس کام کی ناکامی سے ڈیٹا کی ترتیب میں مسائل پیدا ہوسکتے ہیں۔ ڈیٹا اسٹوریج اور ایونٹس کو ایک ساتھ منظم کرنے کے لئے، آپ کو ڈیٹا اسٹوریج کے استعمال ہونے والے ڈیٹا بیس میں پیغامات شائع کرنے یا دو مرحلہ تصدیق (2PC) کا نظام نافذ کرنا ہوگا۔ اگر آپ پیغام بروکر کو ڈیٹا بیس تک تبدیل نہیں کرنا چاہتے اور نہ ہی چکر کھولنا چاہتے ہیں، تو آپ Watermill کے Forwarder کمپوننٹ کا استعمال کرکے اپنے کام کو آسان بنا سکتے ہیں۔

فوروارڈر کمپوننٹ

آپ فوروارڈر کو ایک بیک گراونڈ ڈیمن کی طرح سوچ سکتے ہیں جو پیغامات کے شائع ہونے کا انتظار کرتا ہے اور یہ یقینی بناتا ہے کہ وہ آخر کار پیغام بروکر تک پہنچ جائیں۔

فوروارڈر کو عام اور شفاف بنانے کے لئے، یہ ایک مخصوص موضوع پر وسیط ڈیٹا بیس پر سنتا ہے اور ایک مزین Forwarder publisher کا استعمال کرتا ہے تاکہ مشمولہ پیغامات بھیجے جائیں۔ فوروارڈر انہیں کھولتا ہے اور انہیں مخصوص موضوع پر بھیجتا ہے جو پیغام بروکر پر مشخص کیا گیا ہے۔

مثال

چلیں مندرجہ ذیل مثال کو مد نظر رکھتے ہیں: ایک کمانڈ ہے جو قرعہ اندازی کرنے کے لئے ذمہ دار ہے۔ یہ سسٹم میں رجسٹر شدہ صارف کو بالخصوص انتخاب کرنا ہوگا۔ اس کرتے ہوئے، یہ بھی ڈیسیڈ کرنا چاہیے کہ اس نتیجے کو ڈیٹا بیس میں انٹری کی شکل میں محفوظ کردی گئی ہے جو یکتا قرعہ شناخت کو منتخب صارف کے ID کے ساتھ منسلک کرتی ہے۔ علاوہ ازیں، جیسے ہی واقعہ محور سسٹم ہے، یہ بھی ایک LotteryConcluded واقعہ شائع کرنا چاہیے تاکہ دوسرے حصوں کا مناسب طریقہ عمل ڈال سکیں۔ بالکل ٹھیک - قرعہ جیتنے والے کو انعام بھیجنے کا ذمہ دار کمپوننٹ ہوگا۔ یہ LotteryConcluded واقعہ پر مبنی ہوگا اور ڈیٹا بیس انٹری کے ساتھ بھی دلالت کرے گا جس سے قرعہ شناخت کی منسلکت کا ID پتا کر سکتا ہے۔

ہماری صورتحال میں، ڈیٹا بیس مائی ایس کیو اور میسج بروکر گوگل پب/سب ہے، لیکن یہ کسی بھی دوسری دو تکنیک کا بھی ممکن ہوسکتا ہے۔

ایسی کمانڈ کو نافذ کرتے وقت، مختلف روایات اپنائے جاسکتے ہیں۔ آئیں آگے جاکر تین ممکنہ کوششیں پیش کریں اور ان کی کمیونالیات کو زیر علم لائیں۔

پہلے واقعہ شائع کریں، پھر ڈیٹا سٹور کریں

اس ترتیب کے تحت، کمانڈ پہلے ایک واقعہ شائع کرتی ہے پھر ڈیٹا کو سٹور کرتی ہے. اگرچہ یہ ترکیب زیادہ تر صورتوں میں صحیح کام کرے گی، مگر ہمارے لیے پتہ کرنا ضروری ہے کہ اس میں کونسے مسائل آسکتے ہیں۔

کمانڈ کو تین بنیادی کارروائیاں انجام دینے کی ضرورت ہوتی ہے:

1. ایک رینڈم صارف A کو وائنر کے طور پر منتخب کریں۔
2. LotteryConcluded واقعہ شائع کریں تاکہ مطلوبہ لاٹری B ختم ہو گئی ہے کی اطلاع دی جائے۔
3. ڈیٹا بیس میں محفوظ کریں کہ لاٹری B کو صارف A نے جیت لیا ہے۔

ہر ایک قدم کو کمیابی کے خطرے سے دو ٹوٹنے کا سامنا ہے جو ہماری کمانڈ کی روانی کو برباد کر سکتا ہے۔ اگر پہلا قدم ناکام ہو جائے تو اس کے نتائج کثرت سے سخت نہیں ہوں گے - ہمیں صرف ایک خطا واپس کرنی ہوگی اور تمام کمانڈ کو ناکام سمجھا جائے گا۔ کوئی ڈیٹا ذخیرہ نہیں ہوگا، اور کوئی واقعہ شائع نہیں ہوگا۔ ہم بس کمانڈ کو دوبارہ چلاسکتے ہیں۔

اگر دوسرا قدم ناکام ہو تو ہم نے ابھی تک واقعہ شائع نہیں کیا ہے یا ڈیٹا بیس میں ڈیٹا سٹور نہیں کیا ہے۔ ہم کمانڈ کو دوبارہ چلاسکتے ہیں اور دوبارہ کوشش کرسکتے ہیں۔

سب سے دلچسپ صورتحال یہ ہے کہ تیسرا قدم ناکام ہو تو۔ دوسرے قدم کے بعد، ہم نے پہلے ہی واقعہ شائع کر دیا ہے، مگر انتہائی نتیجہ میں کوئی ڈیٹا ڈیٹا بیس میں محفوظ نہیں ہوگا۔ دوسرے اجزاء کو سگنل ملے گا کہ لاٹری ختم ہوگئی ہے، مگر واقعہ میں بھیجی گئی لاٹری آئی ڈی کے ساتھ کوئی وائنر منسلک نہیں ہوگا۔ انہیں وائنر کی تصدیق نہیں کرنے سکیں گے، لہذا ان کی کارروائیوں کو بھی ناکام سمجھا جانا چاہیے۔

ہم اس صورتحال سے باہر نکل سکتے ہیں، لیکن اس کے لئے کچھ دستی سرگرمیاں شامل ہو سکتی ہیں، جیسے واقعہ سے لاٹری آئی ڈی استعمال کر کے کمانڈ کو دوبارہ چلانا۔

مکمل ماخذ کوڈ: github.com/ThreeDotsLabs/watermill/_examples/real-world-examples/transactional-events-forwarder/main.go

// ...
// 1. پہلے Google Cloud Pub/Sub میں واقعہ شائع کریں، پھر MySQL میں ڈیٹا محفوظ کریں۔
func publishEventAndPersistData(lotteryID int, pickedUser string, logger watermill.LoggerAdapter) error {
	publisher, err := googlecloud.NewPublisher(
		googlecloud.PublisherConfig{
			ProjectID: projectID,
		},
		logger,
	)
	if err != nil {
		return err
	}
	event := LotteryConcludedEvent{LotteryID: lotteryID}
	payload, err := json.Marshal(event)
	if err != nil {
		return err
	}
	err = publisher.Publish(googleCloudEventTopic, message.NewMessage(watermill.NewULID(), payload))
	if err != nil {
		return err
	}

	// اگر کوئی خرابی ہو، تو واقعہ بھیج دیا گیا ہوگا مگر ڈیٹا ابھی تک محفوظ نہیں ہوا ہوگا۔
	if err = simulateError(); err != nil {
		logger.Error("ڈیٹا محفوظ کرنے میں ناکامی", err, nil)
		return err
	}

	_, err = db.Exec(`INSERT INTO lotteries (lottery_id, winner) VALUES(?, ?)`, lotteryID, pickedUser)
	if err != nil {
		return err
	}
// ...

پہلے ڈیٹا ذخیرہ کریں، پھر واقعہ شائع کریں

دوسرے رویہ میں، ہم کوشش کریں گے کہ پہلے استعمال کرنے کی غلطیوں کو دور کریں۔ ڈیٹا بیس میں حالت درست طریقے سے قائم نہیں کی گئی ہو اور واقعات شائع نہیں ہوئے ہیں، تو ہم احتیاط کے طور پر باہری جزویں کو ناکامی کی صورت میں نہیں، روکنے کیلئے، ہم کارروائیوں کے ترتیب کو منسلک کریں گے جیسا کہ مندرجہ ذیل ہے:

1. صورتحال کو بہتر کرنے کے لئے صارف A کو قرعہ B کے برآمد کرنے والا منتخب کریں۔
2. دیتا بیس میں یہ محفوظ کریں کہ قرعہ B کو صارف A نے جیتا ہے۔
3. LotteryConcluded واقعہ منظر عام پر لانے کے لئے جاری کریں، جس میں بتایا گیا ہے کہ قرعہ B ختم ہو گیا ہے۔

پہلے رویے کی طرح، اگر پہلی دو کارروائیاں ناکام ہوں، تو ہمارے پاس کوئی اثرات نہیں ہیں۔ تیسرے کارروائی کی ناکامی کی صورت میں، ہمارا ڈیٹا ڈیٹا بیس میں قائم ہو جائے گا، مگر کوئی واقعہ شائع نہیں ہو گا۔ اس صورت میں، ہم برائے منظور سسٹم کردار کو دیکھتے ہوئے، ہمارے جیتنے والے کو انعام حاصل نہیں ہوگا کیونکہ کوئی واقعہ اس عمل کے ذمہ دار جزو میں نہیں پہنچتا ہے۔

اس مسئلے کو دستی عمل سے بھی حل کیا جا سکتا ہے، مثلاً، یعنی، واقعہ دستی طور پر شائع کرنا۔ مگر ہم اچھی طرح سے کرسکتے ہیں۔

مکمل سورس کوڈ: github.com/ThreeDotsLabs/watermill/_examples/real-world-examples/transactional-events-forwarder/main.go

// 2. پہلے ڈیٹا کو مائی کوئل میں محفوظ کریں، پھر سیدھے گوگل کلاؤڈ پب/سب کو واقعہ شائع کریں۔
func persistDataAndPublishEvent(lotteryID int, pickedUser string, logger watermill.LoggerAdapter) error {
	_, err := db.Exec(`INSERT INTO lotteries (lottery_id, winner) VALUES(?, ?)`, lotteryID, pickedUser)
	if err != nil {
		return err
	}

	var publisher message.Publisher
	publisher, err = googlecloud.NewPublisher(
		googlecloud.PublisherConfig{
			ProjectID: projectID,
		},
		logger,
	)
	if err != nil {
		return err
	}

	event := LotteryConcludedEvent{LotteryID: lotteryID}
	payload, err := json.Marshal(event)
	if err != nil {
		return err
	}

// اگر یہ ناکام ہو جاتا ہے، تو ہمارا ڈیٹا پہلے ہی محفوظ ہو گیا ہے، لیکن کوئی واقعہ شائع نہیں ہوا۔
	if err = simulateError(); err != nil {
		logger.Error("واقعہ شائع کرنے میں ناکام ہوگیا", err, nil)
		return err
	}

	err = publisher.Publish(googleCloudEventTopic, message.NewMessage(watermill.NewULID(), payload))
	if err != nil {
		return err
	}
// ...

ٹرانزیکشن میں ڈیٹا ذخیرہ کرنا اور واقعات شائع کرنا

تصور کریں کہ ہمارا کمانڈ دوسرا اور تیسرا کام ایک ساتھ کر سکتا ہے۔ یہ ایک ہی وقت میں منظور ہو جائیں گے، یعنی اگر ایک کام ناکام ہوتا ہے تو دوسرا کام کامیاب نہیں ہو سکتا۔ یہ اہم ہے کہ موجودہ اکثر ڈیٹا بیس میں پیش کردہ ٹرانزیکشن کی میکانیزم کا فائدہ اٹھایا جا سکتا ہے۔ ہمارے مثال میں استعمال کیا گیا MySQL ایک ایسا ہی ڈیٹا بیس ہے۔

ڈیٹا کو ذخیرہ کرنے اور واقعات کو ایک ہی ٹرانزیکشن میں شائع کرنے کیلئے ، ہمیں یہ ممکن بنانا ہو گا کہ ہم پیغامات کو MySQL میں شائع کریں۔ چونکہ ہم نہیں چاہتے کہ پورے نظام میں یہ تبدیل کیا جائے کہ ہمارے میسج بروکر کو MySQL کی طرف سے سپورٹ کرنا ہو گا، اس لئے ہمیں اس مقصد کو حاصل کرنے کے دوسرے طریقے تلاش کرنے ہوں گے۔

ایسی خوش خبری ہے: Watermill تمام ضروری آلات فراہم کرتا ہے! اگر آپ MySQL ، PostgreSQL (یا کسی دوسرے SQL ڈیٹا بیس) ، Firestore یا Bolt کی طرح کے ڈیٹا بیس استعمال کر رہے ہیں، تو آپ ان میں پیغامات کو شائع کر سکتے ہیں۔ یہ فارورڈر کمپوننٹ آپ کو ان تمام پیغامات کا انتخاب کرنے میں مدد فراہم کرے گا جو آپ ڈیٹا بیس میں شائع کرتے ہیں اور انہیں آپکے میسج بروکر کو فارورڈ کرے گا۔

آپ کو یہ یقینی بنانا ہو گا کہ:

1. آپکا کمانڈ ایک ڈیٹا بیس ٹرانزیکشن کے ساتھ کام کرنے والے پبلشر کا استعمال کرتا ہے (مثلاً SQL, Firestore, Bolt).
2. فارورڈر کمپوننٹ چل رہا ہے، ایک ڈیٹا بیس سبسکرائیبر اور ایک میسج بروکر پبلشر کا استعمال کرتے ہوء۔

اس مقام میں کمانڈ شاید یوں نظر آئے:

// ...
// ایک ٹرانزیکشن کے ساتھ MySQL میں ڈیٹا مستقل اور Google Cloud Pub/Sub میں واقعہ شائع کرنا۔
func persistDataAndPublishEventInTransaction(lotteryID int, pickedUser string, logger watermill.LoggerAdapter) error {
	tx, err := db.Begin()
	if err != nil {
		return err
	}

	defer func() {
		if err == nil {
			tx.Commit()
		} else {
			logger.Info("تراکشن کو ایک خرابی کی وجہ سے واپس لے جانے کے بعد", watermill.LogFields{"error": err.Error()})
			// خرابی کی صورت میں، MySQL ٹرانزیکشن رول بیک کی وجہ سے، ہم زبردستی سے یہ یقینی بنا سکتے ہیں کہ مناسب حالات واقع نہیں ہوں:
			// - واقعہ شائع ہو گیا لیکن ڈیٹا محفوظ نہیں ہوا ہے،
			// - ڈیٹا محفوظ ہوا ہے لیکن واقعہ شائع نہیں ہوا ہے۔
			tx.Rollback()
		}
	}()

	_, err = tx.Exec(`INSERT INTO lotteries (lottery_id, winner) VALUES(?, ?)`, lotteryID, pickedUser)
	if err != nil {
		return err
	}

	var publisher message.Publisher
	publisher, err = sql.NewPublisher(
		tx,
		sql.PublisherConfig{
			SchemaAdapter: sql.DefaultMySQLSchema{},
		},
		logger,
	)
	if err != nil {
		return err
	}

	// پبلشر کو ایک فارورڈر کمپوننٹ کے ذریعہ سمجھا جانے والے پ�...ے سے سنوار کر سجانا۔
	publisher = forwarder.NewPublisher(publisher, forwarder.PublisherConfig{
		ForwarderTopic: forwarderSQLTopic,
	})

	// قرعہ کھیلنے کا اعلان شائع کریں۔ براہ کرم نوٹ کریں کہ ہم یہاں Google Cloud ٹاپک کو استعمال کر کے مناظر بطلان واقعہ شائع کرتے ہیں۔
	event := LotteryConcludedEvent{LotteryID: lotteryID}
	payload, err := json.Marshal(event)
	if err != nil {
		return err
	}

	err = publisher.Publish(googleCloudEventTopic, message.NewMessage(watermill.NewULID(), payload))
	if err != nil {
		return err
	}

	return nil
// ...

اگر آپ اس مثال کے بارے میں مزید جاننا چاہتے ہیں تو براہ کرم اس کے تنصیب کو [yahan](https://github.com/ThreeDotsLabs/watermill/tree/master/_examples/real-world-examples/transactional-events-forwarder/main.go) دیکھیں۔