MessagesHandler.edit_message() - Code Metrics - Inspection of "More tornado complexity refactoring" - Deathangel908/pychat - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 8205ad...0b3f02 )

by Andrew

created 2017-07-02 20:30 UTC

MessagesHandler.edit_message() A

↳ Parent: MessagesHandler

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	4
Bugs	1	Features	0

Metric	Value
cc	2
c	4
b	1
f	0
dl	0
loc	16
rs	9.4285

import json
import logging

from django.core.exceptions import ValidationError
from django.db.models import Q
from tornado.gen import engine, Task
from tornadoredis import Client

from chat.log_filters import id_generator
from chat.models import Message, Room, RoomUsers
from chat.py2_3 import str_type
from chat.settings import ALL_ROOM_ID, SELECT_SELF_ROOM, TORNADO_REDIS_PORT, WEBRTC_CONNECTION
from chat.tornado.constants import VarNames, HandlerNames, Actions, RedisPrefix, WebRtcRedisStates
from chat.tornado.image_utils import process_images, prepare_img, save_images, get_message_images
from chat.tornado.message_creator import WebRtcMessageCreator
from chat.utils import get_max_key, execute_query, do_db, update_room, create_room_users, validate_edit_message, \
	get_or_create_room

parent_logger = logging.getLogger(__name__)
base_logger = logging.LoggerAdapter(parent_logger, {
	'id': 0,
	'ip': '000.000.000.000'
})

# TODO https://github.com/leporo/tornado-redis#connection-pool-support
# CONNECTION_POOL = tornadoredis.ConnectionPool(
# max_connections=500,
# wait_for_available=True)


class MessagesHandler(WebRtcMessageCreator):

	def __init__(self, *args, **kwargs):
		self.closed_channels = None
		self.parsable_prefix = 'p'
		super(MessagesHandler, self).__init__(*args, **kwargs)
		self.webrtc_ids = {}
		self.ip = None
		from chat import global_redis
		self.async_redis_publisher = global_redis.async_redis_publisher
		self.sync_redis = global_redis.sync_redis
		self.channels = []
		self._logger = None
		self.async_redis = Client(port=TORNADO_REDIS_PORT)
		self.patch_tornadoredis()
		self.pre_process_message = {
			Actions.GET_MESSAGES: self.process_get_messages,
			Actions.SEND_MESSAGE: self.process_send_message,
			Actions.WEBRTC: self.proxy_webrtc,
			Actions.CLOSE_FILE_CONNECTION: self.close_file_connection,
			Actions.CLOSE_CALL_CONNECTION: self.close_call_connection,
			Actions.CANCEL_CALL_CONNECTION: self.cancel_call_connection,
			Actions.ACCEPT_CALL: self.accept_call,
			Actions.ACCEPT_FILE: self.accept_file,
			Actions.CREATE_DIRECT_CHANNEL: self.create_user_channel,
			Actions.DELETE_ROOM: self.delete_channel,
			Actions.EDIT_MESSAGE: self.edit_message,
			Actions.CREATE_ROOM_CHANNEL: self.create_new_room,
			Actions.INVITE_USER: self.invite_user,
			Actions.OFFER_FILE_CONNECTION: self.offer_webrtc_connection,
			Actions.OFFER_CALL_CONNECTION: self.offer_webrtc_connection,
			Actions.REPLY_FILE_CONNECTION: self.reply_file_connection,
			Actions.REPLY_CALL_CONNECTION: self.reply_call_connection,
		}
		self.post_process_message = {
			Actions.CREATE_DIRECT_CHANNEL: self.send_client_new_channel,
			Actions.CREATE_ROOM_CHANNEL: self.send_client_new_channel,
			Actions.DELETE_ROOM: self.send_client_delete_channel,
			Actions.INVITE_USER: self.send_client_new_channel,
			Actions.OFFER_FILE_CONNECTION: self.set_opponent_call_channel,
			Actions.OFFER_CALL_CONNECTION: self.set_opponent_call_channel
		}

	def patch_tornadoredis(self):  # TODO remove this
		fabric = type(self.async_redis.connection.readline)
		self.async_redis.connection.old_read = self.async_redis.connection.readline

		def new_read(new_self, callback=None):
			try:
				return new_self.old_read(callback=callback)
			except Exception as e:
				current_online = self.get_online_from_redis(RedisPrefix.DEFAULT_CHANNEL)
				self.logger.error(e)
				self.logger.error(
					"Exception info: "
					"self.id: %s ;;; "
					"self.connected = '%s';;; "
					"Redis default channel online = '%s';;; "
					"self.channels = '%s';;; "
					"self.closed_channels  = '%s';;;",
					self.id, self.connected, current_online, self.channels, self.closed_channels
				)
				raise e

		self.async_redis.connection.readline = fabric(new_read, self.async_redis.connection)

	@property
	def connected(self):
		raise NotImplemented

	@connected.setter
	def connected(self, value):
		raise NotImplemented

	@engine
	def listen(self, channels):
		yield Task(
			self.async_redis.subscribe, channels)
		self.async_redis.listen(self.pub_sub_message)

	@property
	def logger(self):
		return self._logger if self._logger else base_logger

	@engine
	def add_channel(self, channel):
		self.channels.append(channel)
		yield Task(
			self.async_redis.subscribe, (channel,))

	@staticmethod
	def evaluate(query_set):
		do_db(len, query_set)
		return query_set

	def get_online_from_redis(self, channel):
		return self.get_online_and_status_from_redis(channel)[1]

	def get_online_and_status_from_redis(self, channel):
		"""
		:rtype : (bool, list)
		"""
		online = self.sync_redis.ssmembers(channel)
		self.logger.debug('!! channel %s redis online: %s', channel, online)
		return self.parse_redis_online(online) if online else (False, [])

	def parse_redis_online(self, online):
		"""
		:rtype : (bool, list)
		"""
		result = set()
		user_is_online = False
		for decoded in online:  # py2 iteritems
			# : char specified in cookies_middleware.py.create_id
			user_id = int(decoded.split(':')[0])
			if user_id == self.user_id and decoded != self.id:
				user_is_online = True
			result.add(user_id)
		return user_is_online, list(result)

	def add_online_user(self, room_id, offline_messages=None):
		"""
		adds to redis
		online_users = { connection_hash1 = stored_redis_user1, connection_hash_2 = stored_redis_user2 }
		:return:
		"""
		self.async_redis_publisher.sadd(room_id, self.id)
		# since we add user to online first, latest trigger will always show correct online
		is_online, online = self.get_online_and_status_from_redis(room_id)
		if is_online:  # Send user names to self
			online_user_names_mes = self.room_online(online, Actions.REFRESH_USER, room_id)
			self.logger.info('!! Second tab, retrieving online for self')
			self.ws_write(online_user_names_mes)
		else:  # if a new tab has been opened
			online.append(self.user_id)
			online_user_names_mes = self.room_online(online, Actions.LOGIN, room_id)
			self.logger.info('!! First tab, sending refresh online for all')
			self.publish(online_user_names_mes, room_id)
			if offline_messages:
				self.ws_write(self.load_offline_message(offline_messages, room_id))

	def publish(self, message, channel, parsable=False):
		jsoned_mess = json.dumps(message)
		self.logger.debug('<%s> %s', channel, jsoned_mess)
		if parsable:
			jsoned_mess = self.encode(jsoned_mess)
		self.async_redis_publisher.publish(channel, jsoned_mess)

	def encode(self, message):
		"""
		Marks message with prefix to specify that
		it should be decoded and proccesed before sending to client
		@param message: message to mark
		@return: marked message
		"""
		return self.parsable_prefix + message

	def remove_parsable_prefix(self, message):
		if message.startswith(self.parsable_prefix):
			return message[1:]

	def pub_sub_message(self, message):
		data = message.body
		if isinstance(data, str_type):  # subscribe event
			prefixless_str = self.remove_parsable_prefix(data)
			if prefixless_str:
				dict_message = json.loads(prefixless_str)
				res = self.post_process_message[dict_message[VarNames.EVENT]](dict_message)
				if not res:
					self.ws_write(prefixless_str)
			else:
				self.ws_write(data)

	def ws_write(self, message):
		raise NotImplementedError('WebSocketHandler implements')

	def process_send_message(self, message):
		"""
		:type message: dict
		"""
		raw_imgs = message.get(VarNames.IMG)
		channel = message[VarNames.CHANNEL]
		message_db = Message(
			sender_id=self.user_id,
			content=message[VarNames.CONTENT],
			symbol=get_max_key(raw_imgs)
		)
		message_db.room_id = channel
		do_db(message_db.save)
		db_images = save_images(raw_imgs, message_db.id)
		prepared_message = self.create_send_message(
			message_db,
			Actions.PRINT_MESSAGE,
			prepare_img(db_images, message_db.id)
		)
		self.publish(prepared_message, channel)

	def close_file_connection(self, in_message):
		connection_id = in_message[VarNames.CONNECTION_ID]
		self_channel_status = self.sync_redis.shget(connection_id, self.id)
		if not self_channel_status:
			raise Exception("Access Denied")
		if self_channel_status != WebRtcRedisStates.CLOSED:
			sender_id = self.sync_redis.shget(WEBRTC_CONNECTION, connection_id)
			if sender_id == self.id:
				self.close_file_sender(connection_id)
			else:
				self.close_file_receiver(connection_id, in_message, sender_id)
			self.async_redis_publisher.hset(connection_id, self.id, WebRtcRedisStates.CLOSED)

	def close_call_connection(self, in_message):
		self.send_call_answer(
			in_message,
			WebRtcRedisStates.CLOSED,
			Actions.CLOSE_CALL_CONNECTION,
			[WebRtcRedisStates.READY, WebRtcRedisStates.RESPONDED],
			HandlerNames.PEER_CONNECTION
		)

	def cancel_call_connection(self, in_message):
		self.send_call_answer(
			in_message,
			WebRtcRedisStates.CLOSED,
			Actions.CANCEL_CALL_CONNECTION,
			[WebRtcRedisStates.OFFERED],
			HandlerNames.WEBRTC_TRANSFER
		)

	def close_file_receiver(self, connection_id, in_message, sender_id):
		sender_status = self.sync_redis.shget(connection_id, sender_id)
		if not sender_status:
			raise Exception("Access denied")
		if sender_status != WebRtcRedisStates.CLOSED:
			in_message[VarNames.WEBRTC_OPPONENT_ID] = self.id
			in_message[VarNames.HANDLER_NAME] = HandlerNames.PEER_CONNECTION
			self.publish(in_message, sender_id)

	def close_file_sender(self, connection_id):
		values = self.sync_redis.shgetall(connection_id)
		del values[self.id]
		message = self.get_close_file_sender_message(connection_id)
		for ws_id in values:
			if values[ws_id] == WebRtcRedisStates.CLOSED:
				continue
			self.publish(message, ws_id)

	def accept_file(self, in_message):
		connection_id = in_message[VarNames.CONNECTION_ID]
		sender_ws_id = self.sync_redis.shget(WEBRTC_CONNECTION, connection_id)
		sender_ws_status = self.sync_redis.shget(connection_id, sender_ws_id)
		self_ws_status = self.sync_redis.shget(connection_id, self.id)
		if sender_ws_status == WebRtcRedisStates.READY and self_ws_status == WebRtcRedisStates.RESPONDED:
			self.async_redis_publisher.hset(connection_id, self.id, WebRtcRedisStates.READY)
			self.publish(self.get_accept_file_message(connection_id), sender_ws_id)
		else:
			raise ValidationError("Invalid channel status")

	# todo
	# we can use channel_status = self.sync_redis.shgetall(connection_id)
	# and then self.async_redis_publisher.hset(connection_id, self.id, WebRtcRedisStates.READY)
	# if we shgetall and only then do async hset
	# we can catch an issue when 2 concurrent users accepted the call
	# but we didn't  send them ACCEPT_CALL as they both were in status 'offered'
	def accept_call(self, in_message):
		connection_id = in_message[VarNames.CONNECTION_ID]
		self_status = self.sync_redis.shget(connection_id, self.id)
		if self_status == WebRtcRedisStates.RESPONDED:
			conn_users = self.sync_redis.shgetall(connection_id)
			self.publish_call_answer(
				conn_users,
				connection_id,
				HandlerNames.WEBRTC_TRANSFER,
				Actions.ACCEPT_CALL,
				WebRtcRedisStates.READY
			)
		else:
			raise ValidationError("Invalid channel status")

	def offer_webrtc_connection(self, in_message):
		room_id = in_message[VarNames.CHANNEL]
		content = in_message.get(VarNames.CONTENT)
		qued_id = in_message[VarNames.WEBRTC_QUED_ID]
		connection_id = id_generator(RedisPrefix.CONNECTION_ID_LENGTH)
		# use list because sets dont have 1st element which is offerer
		self.async_redis_publisher.hset(WEBRTC_CONNECTION, connection_id, self.id)
		self.async_redis_publisher.hset(connection_id, self.id, WebRtcRedisStates.READY)
		opponents_message = self.offer_webrtc(content, connection_id, room_id, in_message[VarNames.EVENT])
		self_message = self.set_connection_id(qued_id, connection_id)
		self.ws_write(self_message)
		self.logger.info('!! Offering a webrtc, connection_id %s', connection_id)
		self.publish(opponents_message, room_id, True)

	def reply_call_connection(self, in_message):
		self.send_call_answer(
			in_message,
			WebRtcRedisStates.RESPONDED,
			Actions.REPLY_CALL_CONNECTION,
			[WebRtcRedisStates.OFFERED],
			HandlerNames.WEBRTC_TRANSFER
		)

	def send_call_answer(self, in_message, status_set, reply_action, allowed_state, message_handler):
		connection_id = in_message[VarNames.CONNECTION_ID]
		conn_users = self.sync_redis.shgetall(connection_id)
		if conn_users[self.id] in allowed_state:
			self.publish_call_answer(conn_users, connection_id, message_handler, reply_action, status_set)
		else:
			raise ValidationError("Invalid channel status.")

	def publish_call_answer(self, conn_users, connection_id, message_handler, reply_action, status_set):
		self.async_redis_publisher.hset(connection_id, self.id, status_set)
		del conn_users[self.id]
		message = self.reply_webrtc(reply_action, connection_id, message_handler)
		for user in conn_users:
			if conn_users[user] != WebRtcRedisStates.CLOSED:
				self.publish(message, user)

	def reply_file_connection(self, in_message):
		connection_id = in_message[VarNames.CONNECTION_ID]
		sender_ws_id = self.sync_redis.shget(WEBRTC_CONNECTION, connection_id)
		sender_ws_status = self.sync_redis.shget(connection_id, sender_ws_id)
		self_ws_status = self.sync_redis.shget(connection_id, self.id)
		if sender_ws_status == WebRtcRedisStates.READY and self_ws_status == WebRtcRedisStates.OFFERED:
			self.async_redis_publisher.hset(connection_id, self.id, WebRtcRedisStates.RESPONDED)
			self.publish(self.reply_webrtc(
				Actions.REPLY_FILE_CONNECTION,
				connection_id,
				HandlerNames.WEBRTC_TRANSFER
			), sender_ws_id)
		else:
			raise ValidationError("Invalid channel status.")

	def proxy_webrtc(self, in_message):
		"""
		:type in_message: dict
		"""
		connection_id = in_message[VarNames.CONNECTION_ID]
		channel = in_message.get(VarNames.WEBRTC_OPPONENT_ID)
		self_channel_status = self.sync_redis.shget(connection_id, self.id)
		opponent_channel_status = self.sync_redis.shget(connection_id, channel)
		if not (self_channel_status == WebRtcRedisStates.READY and opponent_channel_status == WebRtcRedisStates.READY):
			raise ValidationError('Error in connection status, your status is {} while opponent is {}'.format(
				self_channel_status, opponent_channel_status
			))  # todo receiver should only accept proxy_webrtc from sender, sender can accept all
		# I mean somebody if there're 3 ppl in 1 channel and first is initing transfer to 2nd and 3rd,
		# 2nd guy can fraud 3rd guy webrtc traffic, which is allowed during the call, but not while transering file
		in_message[VarNames.WEBRTC_OPPONENT_ID] = self.id
		in_message[VarNames.HANDLER_NAME] = HandlerNames.PEER_CONNECTION
		self.logger.debug(
			"Forwarding message to channel %s, self %s, other status %s",
			channel,
			self_channel_status,
			opponent_channel_status
		)
		self.publish(in_message, channel)

	def create_new_room(self, message):
		room_name = message[VarNames.ROOM_NAME]
		if not room_name or len(room_name) > 16:
			raise ValidationError('Incorrect room name "{}"'.format(room_name))
		room = Room(name=room_name)
		do_db(room.save)
		RoomUsers(room_id=room.id, user_id=self.user_id).save()
		subscribe_message = self.subscribe_room_channel_message(room.id, room_name)
		self.publish(subscribe_message, self.channel, True)

	def invite_user(self, message):
		room_id = message[VarNames.ROOM_ID]
		user_id = message[VarNames.USER_ID]
		room = get_or_create_room(self.channels, room_id, user_id)
		users_in_room = {
			user.id: RedisPrefix.set_js_user_structure(user.username, user.sex)
			for user in room.users.all()
		}
		self.publish(self.add_user_to_room(room_id, user_id, users_in_room[user_id]), room_id)
		subscribe_message = self.invite_room_channel_message(room_id, user_id, room.name, users_in_room)
		self.publish(subscribe_message, RedisPrefix.generate_user(user_id), True)

	def create_room(self, user_rooms, user_id):
		if self.user_id == user_id:
			room_ids = list([room['room_id'] for room in self.evaluate(user_rooms)])
			query_res = execute_query(SELECT_SELF_ROOM, [room_ids, ])
		else:
			rooms_query = RoomUsers.objects.filter(user_id=user_id, room__in=user_rooms)
			query_res = rooms_query.values('room__id', 'room__disabled')
		try:
			room = do_db(query_res.get)
			room_id = room['room__id']
			update_room(room_id, room['room__disabled'])
		except RoomUsers.DoesNotExist:
			room_id = create_room_users(self.user_id, user_id)
		return room_id

	def create_user_channel(self, message):
		user_id = message[VarNames.USER_ID]
		# get all self private rooms ids
		user_rooms = Room.users.through.objects.filter(user_id=self.user_id, room__name__isnull=True).values('room_id')
		# get private room that contains another user from rooms above
		room_id = self.create_room(user_rooms, user_id)
		subscribe_message = self.subscribe_direct_channel_message(room_id, user_id)
		self.publish(subscribe_message, self.channel, True)
		other_channel = RedisPrefix.generate_user(user_id)
		if self.channel != other_channel:
			self.publish(subscribe_message, other_channel, True)

	def delete_channel(self, message):
		room_id = message[VarNames.ROOM_ID]
		if room_id not in self.channels or room_id == ALL_ROOM_ID:
			raise ValidationError('You are not allowed to exit this room')
		room = do_db(Room.objects.get, id=room_id)
		if room.disabled:
			raise ValidationError('Room is already deleted')
		if room.name is None:  # if private then disable
			room.disabled = True
		else:  # if public -> leave the room, delete the link
			RoomUsers.objects.filter(room_id=room.id, user_id=self.user_id).delete()
			online = self.get_online_from_redis(room_id)
			online.remove(self.user_id)
			self.publish(self.room_online(online, Actions.LOGOUT, room_id), room_id)
		room.save()
		message = self.unsubscribe_direct_message(room_id)
		self.publish(message, room_id, True)

	def edit_message(self, data):
		# ord(next (iter (message['images'])))
		message_id = data[VarNames.MESSAGE_ID]
		message = Message.objects.get(id=message_id)
		validate_edit_message(self.user_id, message)
		message.content = data[VarNames.CONTENT]
		selector = Message.objects.filter(id=message_id)
		if message.content is None:
			action = Actions.DELETE_MESSAGE
			prep_imgs = None
			selector.update(deleted=True)
		else:
			action = Actions.EDIT_MESSAGE
			prep_imgs = process_images(data.get(VarNames.IMG), message)
			selector.update(content=message.content, symbol=message.symbol)
		self.publish(self.create_send_message(message, action, prep_imgs), message.room_id)

	def send_client_new_channel(self, message):
		room_id = message[VarNames.ROOM_ID]
		self.add_channel(room_id)
		self.add_online_user(room_id)

	def set_opponent_call_channel(self, message):
		connection_id = message[VarNames.CONNECTION_ID]
		if message[VarNames.WEBRTC_OPPONENT_ID] == self.id:
			return True
		self.sync_redis.hset(connection_id, self.id, WebRtcRedisStates.OFFERED)

	def send_client_delete_channel(self, message):
		room_id = message[VarNames.ROOM_ID]
		self.async_redis.unsubscribe((room_id,))
		self.async_redis_publisher.hdel(room_id, self.id)
		self.channels.remove(room_id)

	def process_get_messages(self, data):
		"""
		:type data: dict
		"""
		header_id = data.get(VarNames.GET_MESSAGES_HEADER_ID, None)
		count = int(data.get(VarNames.GET_MESSAGES_COUNT, 10))
		room_id = data[VarNames.CHANNEL]
		self.logger.info('!! Fetching %d messages starting from %s', count, header_id)
		if header_id is None:
			messages = Message.objects.filter(Q(room_id=room_id), Q(deleted=False)).order_by('-pk')[:count]
		else:
			messages = Message.objects.filter(Q(id__lt=header_id), Q(room_id=room_id), Q(deleted=False)).order_by('-pk')[:count]
		images = do_db(get_message_images, messages)
		response = self.get_messages(messages, room_id, images)
		self.ws_write(response)


1			import json
2			import logging
3
4			from django.core.exceptions import ValidationError
5			from django.db.models import Q
6			from tornado.gen import engine, Task
7			from tornadoredis import Client
8
9			from chat.log_filters import id_generator
10			from chat.models import Message, Room, RoomUsers
11			from chat.py2_3 import str_type
12			from chat.settings import ALL_ROOM_ID, SELECT_SELF_ROOM, TORNADO_REDIS_PORT, WEBRTC_CONNECTION
13			from chat.tornado.constants import VarNames, HandlerNames, Actions, RedisPrefix, WebRtcRedisStates
14			from chat.tornado.image_utils import process_images, prepare_img, save_images, get_message_images
15			from chat.tornado.message_creator import WebRtcMessageCreator
16			from chat.utils import get_max_key, execute_query, do_db, update_room, create_room_users, validate_edit_message, \
17			get_or_create_room
18
19			parent_logger = logging.getLogger(__name__)
20			base_logger = logging.LoggerAdapter(parent_logger, {
21			'id': 0,
22			'ip': '000.000.000.000'
23			})
24
25			# TODO https://github.com/leporo/tornado-redis#connection-pool-support
26			# CONNECTION_POOL = tornadoredis.ConnectionPool(
27			# max_connections=500,
28			# wait_for_available=True)
29
30
31			class MessagesHandler(WebRtcMessageCreator):
32
33			def __init__(self, args, *kwargs):
34			self.closed_channels = None
35			self.parsable_prefix = 'p'
36			super(MessagesHandler, self).__init__(args, *kwargs)
37			self.webrtc_ids = {}
38			self.ip = None
39			from chat import global_redis
40			self.async_redis_publisher = global_redis.async_redis_publisher
41			self.sync_redis = global_redis.sync_redis
42			self.channels = []
43			self._logger = None
44			self.async_redis = Client(port=TORNADO_REDIS_PORT)
45			self.patch_tornadoredis()
46			self.pre_process_message = {
47			Actions.GET_MESSAGES: self.process_get_messages,
48			Actions.SEND_MESSAGE: self.process_send_message,
49			Actions.WEBRTC: self.proxy_webrtc,
50			Actions.CLOSE_FILE_CONNECTION: self.close_file_connection,
51			Actions.CLOSE_CALL_CONNECTION: self.close_call_connection,
52			Actions.CANCEL_CALL_CONNECTION: self.cancel_call_connection,
53			Actions.ACCEPT_CALL: self.accept_call,
54			Actions.ACCEPT_FILE: self.accept_file,
55			Actions.CREATE_DIRECT_CHANNEL: self.create_user_channel,
56			Actions.DELETE_ROOM: self.delete_channel,
57			Actions.EDIT_MESSAGE: self.edit_message,
58			Actions.CREATE_ROOM_CHANNEL: self.create_new_room,
59			Actions.INVITE_USER: self.invite_user,
60			Actions.OFFER_FILE_CONNECTION: self.offer_webrtc_connection,
61			Actions.OFFER_CALL_CONNECTION: self.offer_webrtc_connection,
62			Actions.REPLY_FILE_CONNECTION: self.reply_file_connection,
63			Actions.REPLY_CALL_CONNECTION: self.reply_call_connection,
64			}
65			self.post_process_message = {
66			Actions.CREATE_DIRECT_CHANNEL: self.send_client_new_channel,
67			Actions.CREATE_ROOM_CHANNEL: self.send_client_new_channel,
68			Actions.DELETE_ROOM: self.send_client_delete_channel,
69			Actions.INVITE_USER: self.send_client_new_channel,
70			Actions.OFFER_FILE_CONNECTION: self.set_opponent_call_channel,
71			Actions.OFFER_CALL_CONNECTION: self.set_opponent_call_channel
72			}
73
74			def patch_tornadoredis(self): # TODO remove this
75			fabric = type(self.async_redis.connection.readline)
76			self.async_redis.connection.old_read = self.async_redis.connection.readline
77
78			def new_read(new_self, callback=None):
79			try:
80			return new_self.old_read(callback=callback)
81			except Exception as e:
82			current_online = self.get_online_from_redis(RedisPrefix.DEFAULT_CHANNEL)
83			self.logger.error(e)
84			self.logger.error(
85			"Exception info: "
86			"self.id: %s ;;; "
87			"self.connected = '%s';;; "
88			"Redis default channel online = '%s';;; "
89			"self.channels = '%s';;; "
90			"self.closed_channels = '%s';;;",
91			self.id, self.connected, current_online, self.channels, self.closed_channels
92			)
93			raise e
94
95			self.async_redis.connection.readline = fabric(new_read, self.async_redis.connection)
96
97			@property
98			def connected(self):
99			raise NotImplemented
100
101			@connected.setter
102			def connected(self, value):
103			raise NotImplemented
104
105			@engine
106			def listen(self, channels):
107			yield Task(
108			self.async_redis.subscribe, channels)
109			self.async_redis.listen(self.pub_sub_message)
110
111			@property
112			def logger(self):
113			return self._logger if self._logger else base_logger
114
115			@engine
116			def add_channel(self, channel):
117			self.channels.append(channel)
118			yield Task(
119			self.async_redis.subscribe, (channel,))
120
121			@staticmethod
122			def evaluate(query_set):
123			do_db(len, query_set)
124			return query_set
125
126			def get_online_from_redis(self, channel):
127			return self.get_online_and_status_from_redis(channel)[1]
128
129			def get_online_and_status_from_redis(self, channel):
130			"""
131			:rtype : (bool, list)
132			"""
133			online = self.sync_redis.ssmembers(channel)
134			self.logger.debug('!! channel %s redis online: %s', channel, online)
135			return self.parse_redis_online(online) if online else (False, [])
136
137			def parse_redis_online(self, online):
138			"""
139			:rtype : (bool, list)
140			"""
141			result = set()
142			user_is_online = False
143			for decoded in online: # py2 iteritems
144			# : char specified in cookies_middleware.py.create_id
145			user_id = int(decoded.split(':')[0])
146			if user_id == self.user_id and decoded != self.id:
147			user_is_online = True
148			result.add(user_id)
149			return user_is_online, list(result)
150
151			def add_online_user(self, room_id, offline_messages=None):
152			"""
153			adds to redis
154			online_users = { connection_hash1 = stored_redis_user1, connection_hash_2 = stored_redis_user2 }
155			:return:
156			"""
157			self.async_redis_publisher.sadd(room_id, self.id)
158			# since we add user to online first, latest trigger will always show correct online
159			is_online, online = self.get_online_and_status_from_redis(room_id)
160			if is_online: # Send user names to self
161			online_user_names_mes = self.room_online(online, Actions.REFRESH_USER, room_id)
162			self.logger.info('!! Second tab, retrieving online for self')
163			self.ws_write(online_user_names_mes)
164			else: # if a new tab has been opened
165			online.append(self.user_id)
166			online_user_names_mes = self.room_online(online, Actions.LOGIN, room_id)
167			self.logger.info('!! First tab, sending refresh online for all')
168			self.publish(online_user_names_mes, room_id)
169			if offline_messages:
170			self.ws_write(self.load_offline_message(offline_messages, room_id))
171
172			def publish(self, message, channel, parsable=False):
173			jsoned_mess = json.dumps(message)
174			self.logger.debug('<%s> %s', channel, jsoned_mess)
175			if parsable:
176			jsoned_mess = self.encode(jsoned_mess)
177			self.async_redis_publisher.publish(channel, jsoned_mess)
178
179			def encode(self, message):
180			"""
181			Marks message with prefix to specify that
182			it should be decoded and proccesed before sending to client
183			@param message: message to mark
184			@return: marked message
185			"""
186			return self.parsable_prefix + message
187
188			def remove_parsable_prefix(self, message):
189			if message.startswith(self.parsable_prefix):
190			return message[1:]
191
192			def pub_sub_message(self, message):
193			data = message.body
194			if isinstance(data, str_type): # subscribe event
195			prefixless_str = self.remove_parsable_prefix(data)
196			if prefixless_str:
197			dict_message = json.loads(prefixless_str)
198			res = self.post_process_message[dict_message[VarNames.EVENT]](dict_message)
199			if not res:
200			self.ws_write(prefixless_str)
201			else:
202			self.ws_write(data)
203
204			def ws_write(self, message):
205			raise NotImplementedError('WebSocketHandler implements')
206
207			def process_send_message(self, message):
208			"""
209			:type message: dict
210			"""
211			raw_imgs = message.get(VarNames.IMG)
212			channel = message[VarNames.CHANNEL]
213			message_db = Message(
214			sender_id=self.user_id,
215			content=message[VarNames.CONTENT],
216			symbol=get_max_key(raw_imgs)
217			)
218			message_db.room_id = channel
219			do_db(message_db.save)
220			db_images = save_images(raw_imgs, message_db.id)
221			prepared_message = self.create_send_message(
222			message_db,
223			Actions.PRINT_MESSAGE,
224			prepare_img(db_images, message_db.id)
225			)
226			self.publish(prepared_message, channel)
227
228			def close_file_connection(self, in_message):
229			connection_id = in_message[VarNames.CONNECTION_ID]
230			self_channel_status = self.sync_redis.shget(connection_id, self.id)
231			if not self_channel_status:
232			raise Exception("Access Denied")
233			if self_channel_status != WebRtcRedisStates.CLOSED:
234			sender_id = self.sync_redis.shget(WEBRTC_CONNECTION, connection_id)
235			if sender_id == self.id:
236			self.close_file_sender(connection_id)
237			else:
238			self.close_file_receiver(connection_id, in_message, sender_id)
239			self.async_redis_publisher.hset(connection_id, self.id, WebRtcRedisStates.CLOSED)
240
241			def close_call_connection(self, in_message):
242			self.send_call_answer(
243			in_message,
244			WebRtcRedisStates.CLOSED,
245			Actions.CLOSE_CALL_CONNECTION,
246			[WebRtcRedisStates.READY, WebRtcRedisStates.RESPONDED],
247			HandlerNames.PEER_CONNECTION
248			)
249
250			def cancel_call_connection(self, in_message):
251			self.send_call_answer(
252			in_message,
253			WebRtcRedisStates.CLOSED,
254			Actions.CANCEL_CALL_CONNECTION,
255			[WebRtcRedisStates.OFFERED],
256			HandlerNames.WEBRTC_TRANSFER
257			)
258
259			def close_file_receiver(self, connection_id, in_message, sender_id):
260			sender_status = self.sync_redis.shget(connection_id, sender_id)
261			if not sender_status:
262			raise Exception("Access denied")
263			if sender_status != WebRtcRedisStates.CLOSED:
264			in_message[VarNames.WEBRTC_OPPONENT_ID] = self.id
265			in_message[VarNames.HANDLER_NAME] = HandlerNames.PEER_CONNECTION
266			self.publish(in_message, sender_id)
267
268			def close_file_sender(self, connection_id):
269			values = self.sync_redis.shgetall(connection_id)
270			del values[self.id]
271			message = self.get_close_file_sender_message(connection_id)
272			for ws_id in values:
273			if values[ws_id] == WebRtcRedisStates.CLOSED:
274			continue
275			self.publish(message, ws_id)
276
277			def accept_file(self, in_message):
278			connection_id = in_message[VarNames.CONNECTION_ID]
279			sender_ws_id = self.sync_redis.shget(WEBRTC_CONNECTION, connection_id)
280			sender_ws_status = self.sync_redis.shget(connection_id, sender_ws_id)
281			self_ws_status = self.sync_redis.shget(connection_id, self.id)
282			if sender_ws_status == WebRtcRedisStates.READY and self_ws_status == WebRtcRedisStates.RESPONDED:
283			self.async_redis_publisher.hset(connection_id, self.id, WebRtcRedisStates.READY)
284			self.publish(self.get_accept_file_message(connection_id), sender_ws_id)
285			else:
286			raise ValidationError("Invalid channel status")
287
288			# todo
289			# we can use channel_status = self.sync_redis.shgetall(connection_id)
290			# and then self.async_redis_publisher.hset(connection_id, self.id, WebRtcRedisStates.READY)
291			# if we shgetall and only then do async hset
292			# we can catch an issue when 2 concurrent users accepted the call
293			# but we didn't send them ACCEPT_CALL as they both were in status 'offered'
294			def accept_call(self, in_message):
295			connection_id = in_message[VarNames.CONNECTION_ID]
296			self_status = self.sync_redis.shget(connection_id, self.id)
297			if self_status == WebRtcRedisStates.RESPONDED:
298			conn_users = self.sync_redis.shgetall(connection_id)
299			self.publish_call_answer(
300			conn_users,
301			connection_id,
302			HandlerNames.WEBRTC_TRANSFER,
303			Actions.ACCEPT_CALL,
304			WebRtcRedisStates.READY
305			)
306			else:
307			raise ValidationError("Invalid channel status")
308
309			def offer_webrtc_connection(self, in_message):
310			room_id = in_message[VarNames.CHANNEL]
311			content = in_message.get(VarNames.CONTENT)
312			qued_id = in_message[VarNames.WEBRTC_QUED_ID]
313			connection_id = id_generator(RedisPrefix.CONNECTION_ID_LENGTH)
314			# use list because sets dont have 1st element which is offerer
315			self.async_redis_publisher.hset(WEBRTC_CONNECTION, connection_id, self.id)
316			self.async_redis_publisher.hset(connection_id, self.id, WebRtcRedisStates.READY)
317			opponents_message = self.offer_webrtc(content, connection_id, room_id, in_message[VarNames.EVENT])
318			self_message = self.set_connection_id(qued_id, connection_id)
319			self.ws_write(self_message)
320			self.logger.info('!! Offering a webrtc, connection_id %s', connection_id)
321			self.publish(opponents_message, room_id, True)
322
323			def reply_call_connection(self, in_message):
324			self.send_call_answer(
325			in_message,
326			WebRtcRedisStates.RESPONDED,
327			Actions.REPLY_CALL_CONNECTION,
328			[WebRtcRedisStates.OFFERED],
329			HandlerNames.WEBRTC_TRANSFER
330			)
331
332			def send_call_answer(self, in_message, status_set, reply_action, allowed_state, message_handler):
333			connection_id = in_message[VarNames.CONNECTION_ID]
334			conn_users = self.sync_redis.shgetall(connection_id)
335			if conn_users[self.id] in allowed_state:
336			self.publish_call_answer(conn_users, connection_id, message_handler, reply_action, status_set)
337			else:
338			raise ValidationError("Invalid channel status.")
339
340			def publish_call_answer(self, conn_users, connection_id, message_handler, reply_action, status_set):
341			self.async_redis_publisher.hset(connection_id, self.id, status_set)
342			del conn_users[self.id]
343			message = self.reply_webrtc(reply_action, connection_id, message_handler)
344			for user in conn_users:
345			if conn_users[user] != WebRtcRedisStates.CLOSED:
346			self.publish(message, user)
347
348			def reply_file_connection(self, in_message):
349			connection_id = in_message[VarNames.CONNECTION_ID]
350			sender_ws_id = self.sync_redis.shget(WEBRTC_CONNECTION, connection_id)
351			sender_ws_status = self.sync_redis.shget(connection_id, sender_ws_id)
352			self_ws_status = self.sync_redis.shget(connection_id, self.id)
353			if sender_ws_status == WebRtcRedisStates.READY and self_ws_status == WebRtcRedisStates.OFFERED:
354			self.async_redis_publisher.hset(connection_id, self.id, WebRtcRedisStates.RESPONDED)
355			self.publish(self.reply_webrtc(
356			Actions.REPLY_FILE_CONNECTION,
357			connection_id,
358			HandlerNames.WEBRTC_TRANSFER
359			), sender_ws_id)
360			else:
361			raise ValidationError("Invalid channel status.")
362
363			def proxy_webrtc(self, in_message):
364			"""
365			:type in_message: dict
366			"""
367			connection_id = in_message[VarNames.CONNECTION_ID]
368			channel = in_message.get(VarNames.WEBRTC_OPPONENT_ID)
369			self_channel_status = self.sync_redis.shget(connection_id, self.id)
370			opponent_channel_status = self.sync_redis.shget(connection_id, channel)
371			if not (self_channel_status == WebRtcRedisStates.READY and opponent_channel_status == WebRtcRedisStates.READY):
372			raise ValidationError('Error in connection status, your status is {} while opponent is {}'.format(
373			self_channel_status, opponent_channel_status
374			)) # todo receiver should only accept proxy_webrtc from sender, sender can accept all
375			# I mean somebody if there're 3 ppl in 1 channel and first is initing transfer to 2nd and 3rd,
376			# 2nd guy can fraud 3rd guy webrtc traffic, which is allowed during the call, but not while transering file
377			in_message[VarNames.WEBRTC_OPPONENT_ID] = self.id
378			in_message[VarNames.HANDLER_NAME] = HandlerNames.PEER_CONNECTION
379			self.logger.debug(
380			"Forwarding message to channel %s, self %s, other status %s",
381			channel,
382			self_channel_status,
383			opponent_channel_status
384			)
385			self.publish(in_message, channel)
386
387			def create_new_room(self, message):
388			room_name = message[VarNames.ROOM_NAME]
389			if not room_name or len(room_name) > 16:
390			raise ValidationError('Incorrect room name "{}"'.format(room_name))
391			room = Room(name=room_name)
392			do_db(room.save)
393			RoomUsers(room_id=room.id, user_id=self.user_id).save()
394			subscribe_message = self.subscribe_room_channel_message(room.id, room_name)
395			self.publish(subscribe_message, self.channel, True)
396
397			def invite_user(self, message):
398			room_id = message[VarNames.ROOM_ID]
399			user_id = message[VarNames.USER_ID]
400			room = get_or_create_room(self.channels, room_id, user_id)
401			users_in_room = {
402			user.id: RedisPrefix.set_js_user_structure(user.username, user.sex)
403			for user in room.users.all()
404			}
405			self.publish(self.add_user_to_room(room_id, user_id, users_in_room[user_id]), room_id)
406			subscribe_message = self.invite_room_channel_message(room_id, user_id, room.name, users_in_room)
407			self.publish(subscribe_message, RedisPrefix.generate_user(user_id), True)
408
409			def create_room(self, user_rooms, user_id):
410			if self.user_id == user_id:
411			room_ids = list([room['room_id'] for room in self.evaluate(user_rooms)])
412			query_res = execute_query(SELECT_SELF_ROOM, [room_ids, ])
413			else:
414			rooms_query = RoomUsers.objects.filter(user_id=user_id, room__in=user_rooms)
415			query_res = rooms_query.values('room__id', 'room__disabled')
416			try:
417			room = do_db(query_res.get)
418			room_id = room['room__id']
419			update_room(room_id, room['room__disabled'])
420			except RoomUsers.DoesNotExist:
421			room_id = create_room_users(self.user_id, user_id)
422			return room_id
423
424			def create_user_channel(self, message):
425			user_id = message[VarNames.USER_ID]
426			# get all self private rooms ids
427			user_rooms = Room.users.through.objects.filter(user_id=self.user_id, room__name__isnull=True).values('room_id')
428			# get private room that contains another user from rooms above
429			room_id = self.create_room(user_rooms, user_id)
430			subscribe_message = self.subscribe_direct_channel_message(room_id, user_id)
431			self.publish(subscribe_message, self.channel, True)
432			other_channel = RedisPrefix.generate_user(user_id)
433			if self.channel != other_channel:
434			self.publish(subscribe_message, other_channel, True)
435
436			def delete_channel(self, message):
437			room_id = message[VarNames.ROOM_ID]
438			if room_id not in self.channels or room_id == ALL_ROOM_ID:
439			raise ValidationError('You are not allowed to exit this room')
440			room = do_db(Room.objects.get, id=room_id)
441			if room.disabled:
442			raise ValidationError('Room is already deleted')
443			if room.name is None: # if private then disable
444			room.disabled = True
445			else: # if public -> leave the room, delete the link
446			RoomUsers.objects.filter(room_id=room.id, user_id=self.user_id).delete()
447			online = self.get_online_from_redis(room_id)
448			online.remove(self.user_id)
449			self.publish(self.room_online(online, Actions.LOGOUT, room_id), room_id)
450			room.save()
451			message = self.unsubscribe_direct_message(room_id)
452			self.publish(message, room_id, True)
453
454			def edit_message(self, data):
455			# ord(next (iter (message['images'])))
456			message_id = data[VarNames.MESSAGE_ID]
457			message = Message.objects.get(id=message_id)
458			validate_edit_message(self.user_id, message)
459			message.content = data[VarNames.CONTENT]
460			selector = Message.objects.filter(id=message_id)
461			if message.content is None:
462			action = Actions.DELETE_MESSAGE
463			prep_imgs = None
464			selector.update(deleted=True)
465			else:
466			action = Actions.EDIT_MESSAGE
467			prep_imgs = process_images(data.get(VarNames.IMG), message)
468			selector.update(content=message.content, symbol=message.symbol)
469			self.publish(self.create_send_message(message, action, prep_imgs), message.room_id)
470
471			def send_client_new_channel(self, message):
472			room_id = message[VarNames.ROOM_ID]
473			self.add_channel(room_id)
474			self.add_online_user(room_id)
475
476			def set_opponent_call_channel(self, message):
477			connection_id = message[VarNames.CONNECTION_ID]
478			if message[VarNames.WEBRTC_OPPONENT_ID] == self.id:
479			return True
480			self.sync_redis.hset(connection_id, self.id, WebRtcRedisStates.OFFERED)
481
482			def send_client_delete_channel(self, message):
483			room_id = message[VarNames.ROOM_ID]
484			self.async_redis.unsubscribe((room_id,))
485			self.async_redis_publisher.hdel(room_id, self.id)
486			self.channels.remove(room_id)
487
488			def process_get_messages(self, data):
489			"""
490			:type data: dict
491			"""
492			header_id = data.get(VarNames.GET_MESSAGES_HEADER_ID, None)
493			count = int(data.get(VarNames.GET_MESSAGES_COUNT, 10))
494			room_id = data[VarNames.CHANNEL]
495			self.logger.info('!! Fetching %d messages starting from %s', count, header_id)
496			if header_id is None:
497			messages = Message.objects.filter(Q(room_id=room_id), Q(deleted=False)).order_by('-pk')[:count]
498			else:
499			messages = Message.objects.filter(Q(id__lt=header_id), Q(room_id=room_id), Q(deleted=False)).order_by('-pk')[:count]
500			images = do_db(get_message_images, messages)
501			response = self.get_messages(messages, room_id, images)
502			self.ws_write(response)
503

Deathangel908 / pychat

Push — master ( 8205ad...0b3f02 )

MessagesHandler.edit_message() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like