WebSocket

Narmak ERP использует WebSocket для получения начальных данных и real-time уведомлений.

Backend: Django Channels + Daphne (ASGI)
Frontend: libs/websocket/websocket.service.ts

Архитектура

Конфигурация

// libs/environments/src/environment.ts
export const environment = {
  wsUrl: 'ws://127.0.0.1:8018/test',
  // в production:
  // wsUrl: 'wss://office.narmak.ru/ws/'
}

# settings.py
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("redis", 6379)],
        },
    },
}

ASGI_APPLICATION = "narmak.asgi.application"

Frontend: WebsocketService

Файл: libs/websocket/src/lib/websocket.service.ts

@Injectable({ providedIn: 'root' })
export class WebsocketService {
  private ws: WebSocket | null = null
  private reconnectAttempts = 0
  private readonly MAX_RECONNECTS = 3
  private messages$ = new Subject<any>()

  connect(): void {
    this.ws = new WebSocket(environment.wsUrl)

    this.ws.onopen = () => {
      this.reconnectAttempts = 0
      console.log('WebSocket connected')
      this.sendMessage({ type: 'get_initial_data' })
    }

    this.ws.onmessage = (event) => {
      const data = JSON.parse(event.data)
      this.messages$.next(data)
    }

    this.ws.onclose = () => {
      if (this.reconnectAttempts < this.MAX_RECONNECTS) {
        this.reconnectAttempts++
        setTimeout(() => this.connect(), 3000)
      }
    }

    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error)
    }
  }

  sendMessage(data: object): void {
    if (this.ws?.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(data))
    }
  }

  getInitialData(): Observable<InitialData> {
    return this.messages$.pipe(
      filter(msg => msg.type === 'initial_data'),
      map(msg => msg.data),
      take(1)
    )
  }

  on<T>(type: string): Observable<T> {
    return this.messages$.pipe(
      filter(msg => msg.type === type),
      map(msg => msg.data)
    )
  }
}

Структура сообщений

Клиент → Сервер

// Запрос начальных данных
{ "type": "get_initial_data" }

// Подписка на обновления документа
{ "type": "subscribe_document", "document_id": 123 }

Сервер → Клиент

// Начальные данные при подключении
{
  "type": "initial_data",
  "data": {
    "counterparties": [...],
    "organizations": [...],
    "assortiment": [...],
    "current_user": {...},
    "vocabulary": {...}
  }
}

// Уведомление об изменении документа
{
  "type": "document_updated",
  "data": { "document_id": 123, "status": "confirmed" }
}

// Уведомление о новой задаче
{
  "type": "task_assigned",
  "data": { "task_id": 456, "title": "Проверить отчёт" }
}

Backend: Django Channels Consumer

# narmak/consumers.py
from channels.generic.websocket import AsyncWebsocketConsumer

class NarmakConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        user = self.scope['user']
        if not user.is_authenticated:
            await self.close()
            return

        self.room_name = f"user_{user.id}"
        await self.channel_layer.group_add(
            self.room_name,
            self.channel_name
        )
        await self.accept()

    async def receive(self, text_data):
        data = json.loads(text_data)

        if data['type'] == 'get_initial_data':
            initial_data = await self.get_initial_data()
            await self.send(json.dumps({
                'type': 'initial_data',
                'data': initial_data
            }))

    async def get_initial_data(self):
        """Загрузить начальные данные для пользователя"""
        return {
            'counterparties': await self.get_counterparties(),
            'organizations': await self.get_organizations(),
            'assortiment': await self.get_assortiment(),
        }

Инициализация в приложении

// apps/main-office/src/app/app.component.ts
@Component({...})
export class AppComponent implements OnInit {
  constructor(private initDataService: InitDataRootService) {}

  ngOnInit() {
    // Подключение WebSocket и загрузка начальных данных
    this.initDataService.init()
  }
}

Переподключение

При разрыве соединения сервис автоматически пытается переподключиться:

• 3 попытки с интервалом 3 секунды
• После 3 неудачных попыток — показывается сообщение пользователю
• При восстановлении соединения — повторный запрос get_initial_data

Связанные разделы

• State Management — Akita stores
• Развёртывание — Daphne конфигурация